VOL-3431: Following enhancement/changes are done in this patch
- Process ONUs in bulk rather than serial, significantly improves run time
- Used a separate API to get flow-id. This flow-id is not freed on failure,
as this adds to unnecessary complexity and unwarranted for a test tool .
- Print the total execution time at the end of the test
- Fixed the Dockerfile to not build vendor module at each docker build time
- Introduced new functions to retry scheduler, queue and flow adds on failure,
but these are not currently used
- Add vendor modules to repo just like all other ONF VOLTHA golang projects do
- Tested all three workflows - ATT, DT and TT
- Bump version to 1.1.0
Change-Id: I6102cb206e78ea04b49b7125b101946ca6f36bfb
diff --git a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/backend.go b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/backend.go
new file mode 100644
index 0000000..faa86ed
--- /dev/null
+++ b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/backend.go
@@ -0,0 +1,269 @@
+/*
+ * Copyright 2018-present Open Networking Foundation
+
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+
+ * http://www.apache.org/licenses/LICENSE-2.0
+
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package db
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "strconv"
+ "sync"
+ "time"
+
+ "github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore"
+ "github.com/opencord/voltha-lib-go/v3/pkg/log"
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/status"
+)
+
+const (
+ // Default Minimal Interval for posting alive state of backend kvstore on Liveness Channel
+ DefaultLivenessChannelInterval = time.Second * 30
+)
+
+// Backend structure holds details for accessing the kv store
+type Backend struct {
+ sync.RWMutex
+ Client kvstore.Client
+ StoreType string
+ Host string
+ Port int
+ Timeout int
+ PathPrefix string
+ alive bool // Is this backend connection alive?
+ liveness chan bool // channel to post alive state
+ LivenessChannelInterval time.Duration // regularly push alive state beyond this interval
+ lastLivenessTime time.Time // Instant of last alive state push
+}
+
+// NewBackend creates a new instance of a Backend structure
+func NewBackend(storeType string, host string, port int, timeout int, pathPrefix string) *Backend {
+ var err error
+
+ b := &Backend{
+ StoreType: storeType,
+ Host: host,
+ Port: port,
+ Timeout: timeout,
+ LivenessChannelInterval: DefaultLivenessChannelInterval,
+ PathPrefix: pathPrefix,
+ alive: false, // connection considered down at start
+ }
+
+ address := host + ":" + strconv.Itoa(port)
+ if b.Client, err = b.newClient(address, timeout); err != nil {
+ logger.Errorw("failed-to-create-kv-client",
+ log.Fields{
+ "type": storeType, "host": host, "port": port,
+ "timeout": timeout, "prefix": pathPrefix,
+ "error": err.Error(),
+ })
+ }
+
+ return b
+}
+
+func (b *Backend) newClient(address string, timeout int) (kvstore.Client, error) {
+ switch b.StoreType {
+ case "consul":
+ return kvstore.NewConsulClient(address, timeout)
+ case "etcd":
+ return kvstore.NewEtcdClient(address, timeout)
+ }
+ return nil, errors.New("unsupported-kv-store")
+}
+
+func (b *Backend) makePath(key string) string {
+ path := fmt.Sprintf("%s/%s", b.PathPrefix, key)
+ return path
+}
+
+func (b *Backend) updateLiveness(alive bool) {
+ // Periodically push stream of liveness data to the channel,
+ // so that in a live state, the core does not timeout and
+ // send a forced liveness message. Push alive state if the
+ // last push to channel was beyond livenessChannelInterval
+ if b.liveness != nil {
+
+ if b.alive != alive {
+ logger.Debug("update-liveness-channel-reason-change")
+ b.liveness <- alive
+ b.lastLivenessTime = time.Now()
+ } else if time.Since(b.lastLivenessTime) > b.LivenessChannelInterval {
+ logger.Debug("update-liveness-channel-reason-interval")
+ b.liveness <- alive
+ b.lastLivenessTime = time.Now()
+ }
+ }
+
+ // Emit log message only for alive state change
+ if b.alive != alive {
+ logger.Debugw("change-kvstore-alive-status", log.Fields{"alive": alive})
+ b.alive = alive
+ }
+}
+
+// Perform a dummy Key Lookup on kvstore to test Connection Liveness and
+// post on Liveness channel
+func (b *Backend) PerformLivenessCheck(ctx context.Context) bool {
+ alive := b.Client.IsConnectionUp(ctx)
+ logger.Debugw("kvstore-liveness-check-result", log.Fields{"alive": alive})
+
+ b.updateLiveness(alive)
+ return alive
+}
+
+// Enable the liveness monitor channel. This channel will report
+// a "true" or "false" on every kvstore operation which indicates whether
+// or not the connection is still Live. This channel is then picked up
+// by the service (i.e. rw_core / ro_core) to update readiness status
+// and/or take other actions.
+func (b *Backend) EnableLivenessChannel() chan bool {
+ logger.Debug("enable-kvstore-liveness-channel")
+
+ if b.liveness == nil {
+ logger.Debug("create-kvstore-liveness-channel")
+
+ // Channel size of 10 to avoid any possibility of blocking in Load conditions
+ b.liveness = make(chan bool, 10)
+
+ // Post initial alive state
+ b.liveness <- b.alive
+ b.lastLivenessTime = time.Now()
+ }
+
+ return b.liveness
+}
+
+// Extract Alive status of Kvstore based on type of error
+func (b *Backend) isErrorIndicatingAliveKvstore(err error) bool {
+ // Alive unless observed an error indicating so
+ alive := true
+
+ if err != nil {
+
+ // timeout indicates kvstore not reachable/alive
+ if err == context.DeadlineExceeded {
+ alive = false
+ }
+
+ // Need to analyze client-specific errors based on backend type
+ if b.StoreType == "etcd" {
+
+ // For etcd backend, consider not-alive only for errors indicating
+ // timedout request or unavailable/corrupted cluster. For all remaining
+ // error codes listed in https://godoc.org/google.golang.org/grpc/codes#Code,
+ // we would not infer a not-alive backend because such a error may also
+ // occur due to bad client requests or sequence of operations
+ switch status.Code(err) {
+ case codes.DeadlineExceeded:
+ fallthrough
+ case codes.Unavailable:
+ fallthrough
+ case codes.DataLoss:
+ alive = false
+ }
+
+ //} else {
+ // TODO: Implement for consul backend; would it be needed ever?
+ }
+ }
+
+ return alive
+}
+
+// List retrieves one or more items that match the specified key
+func (b *Backend) List(ctx context.Context, key string) (map[string]*kvstore.KVPair, error) {
+ b.Lock()
+ defer b.Unlock()
+
+ formattedPath := b.makePath(key)
+ logger.Debugw("listing-key", log.Fields{"key": key, "path": formattedPath})
+
+ pair, err := b.Client.List(ctx, formattedPath)
+
+ b.updateLiveness(b.isErrorIndicatingAliveKvstore(err))
+
+ return pair, err
+}
+
+// Get retrieves an item that matches the specified key
+func (b *Backend) Get(ctx context.Context, key string) (*kvstore.KVPair, error) {
+ b.Lock()
+ defer b.Unlock()
+
+ formattedPath := b.makePath(key)
+ logger.Debugw("getting-key", log.Fields{"key": key, "path": formattedPath})
+
+ pair, err := b.Client.Get(ctx, formattedPath)
+
+ b.updateLiveness(b.isErrorIndicatingAliveKvstore(err))
+
+ return pair, err
+}
+
+// Put stores an item value under the specifed key
+func (b *Backend) Put(ctx context.Context, key string, value interface{}) error {
+ b.Lock()
+ defer b.Unlock()
+
+ formattedPath := b.makePath(key)
+ logger.Debugw("putting-key", log.Fields{"key": key, "value": value, "path": formattedPath})
+
+ err := b.Client.Put(ctx, formattedPath, value)
+
+ b.updateLiveness(b.isErrorIndicatingAliveKvstore(err))
+
+ return err
+}
+
+// Delete removes an item under the specified key
+func (b *Backend) Delete(ctx context.Context, key string) error {
+ b.Lock()
+ defer b.Unlock()
+
+ formattedPath := b.makePath(key)
+ logger.Debugw("deleting-key", log.Fields{"key": key, "path": formattedPath})
+
+ err := b.Client.Delete(ctx, formattedPath)
+
+ b.updateLiveness(b.isErrorIndicatingAliveKvstore(err))
+
+ return err
+}
+
+// CreateWatch starts watching events for the specified key
+func (b *Backend) CreateWatch(ctx context.Context, key string, withPrefix bool) chan *kvstore.Event {
+ b.Lock()
+ defer b.Unlock()
+
+ formattedPath := b.makePath(key)
+ logger.Debugw("creating-key-watch", log.Fields{"key": key, "path": formattedPath})
+
+ return b.Client.Watch(ctx, formattedPath, withPrefix)
+}
+
+// DeleteWatch stops watching events for the specified key
+func (b *Backend) DeleteWatch(key string, ch chan *kvstore.Event) {
+ b.Lock()
+ defer b.Unlock()
+
+ formattedPath := b.makePath(key)
+ logger.Debugw("deleting-key-watch", log.Fields{"key": key, "path": formattedPath})
+
+ b.Client.CloseWatch(formattedPath, ch)
+}
diff --git a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/common.go b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/common.go
new file mode 100644
index 0000000..a5a79ae
--- /dev/null
+++ b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/common.go
@@ -0,0 +1,35 @@
+/*
+ * Copyright 2020-present Open Networking Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package db
+
+import (
+ "github.com/opencord/voltha-lib-go/v3/pkg/log"
+)
+
+const (
+ logLevel = log.ErrorLevel
+)
+
+var logger log.Logger
+
+func init() {
+ // Setup this package so that it's log level can be modified at run time
+ var err error
+ logger, err = log.AddPackage(log.JSON, logLevel, log.Fields{"pkg": "db"})
+ if err != nil {
+ panic(err)
+ }
+}
diff --git a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore/client.go b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore/client.go
new file mode 100644
index 0000000..b9cb1ee
--- /dev/null
+++ b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore/client.go
@@ -0,0 +1,90 @@
+/*
+ * Copyright 2018-present Open Networking Foundation
+
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+
+ * http://www.apache.org/licenses/LICENSE-2.0
+
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package kvstore
+
+import "context"
+
+const (
+ // Default timeout in seconds when making a kvstore request
+ defaultKVGetTimeout = 5
+ // Maximum channel buffer between publisher/subscriber goroutines
+ maxClientChannelBufferSize = 10
+)
+
+// These constants represent the event types returned by the KV client
+const (
+ PUT = iota
+ DELETE
+ CONNECTIONDOWN
+ UNKNOWN
+)
+
+// KVPair is a common wrapper for key-value pairs returned from the KV store
+type KVPair struct {
+ Key string
+ Value interface{}
+ Version int64
+ Session string
+ Lease int64
+}
+
+// NewKVPair creates a new KVPair object
+func NewKVPair(key string, value interface{}, session string, lease int64, version int64) *KVPair {
+ kv := new(KVPair)
+ kv.Key = key
+ kv.Value = value
+ kv.Session = session
+ kv.Lease = lease
+ kv.Version = version
+ return kv
+}
+
+// Event is generated by the KV client when a key change is detected
+type Event struct {
+ EventType int
+ Key interface{}
+ Value interface{}
+ Version int64
+}
+
+// NewEvent creates a new Event object
+func NewEvent(eventType int, key interface{}, value interface{}, version int64) *Event {
+ evnt := new(Event)
+ evnt.EventType = eventType
+ evnt.Key = key
+ evnt.Value = value
+ evnt.Version = version
+
+ return evnt
+}
+
+// Client represents the set of APIs a KV Client must implement
+type Client interface {
+ List(ctx context.Context, key string) (map[string]*KVPair, error)
+ Get(ctx context.Context, key string) (*KVPair, error)
+ Put(ctx context.Context, key string, value interface{}) error
+ Delete(ctx context.Context, key string) error
+ Reserve(ctx context.Context, key string, value interface{}, ttl int64) (interface{}, error)
+ ReleaseReservation(ctx context.Context, key string) error
+ ReleaseAllReservations(ctx context.Context) error
+ RenewReservation(ctx context.Context, key string) error
+ Watch(ctx context.Context, key string, withPrefix bool) chan *Event
+ AcquireLock(ctx context.Context, lockName string, timeout int) error
+ ReleaseLock(lockName string) error
+ IsConnectionUp(ctx context.Context) bool // timeout in second
+ CloseWatch(key string, ch chan *Event)
+ Close()
+}
diff --git a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore/common.go b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore/common.go
new file mode 100644
index 0000000..2d2a6a6
--- /dev/null
+++ b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore/common.go
@@ -0,0 +1,35 @@
+/*
+ * Copyright 2020-present Open Networking Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package kvstore
+
+import (
+ "github.com/opencord/voltha-lib-go/v3/pkg/log"
+)
+
+const (
+ logLevel = log.ErrorLevel
+)
+
+var logger log.Logger
+
+func init() {
+ // Setup this package so that it's log level can be modified at run time
+ var err error
+ logger, err = log.AddPackage(log.JSON, logLevel, log.Fields{"pkg": "kvstore"})
+ if err != nil {
+ panic(err)
+ }
+}
diff --git a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore/consulclient.go b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore/consulclient.go
new file mode 100644
index 0000000..bdf2d10
--- /dev/null
+++ b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore/consulclient.go
@@ -0,0 +1,512 @@
+/*
+ * Copyright 2018-present Open Networking Foundation
+
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+
+ * http://www.apache.org/licenses/LICENSE-2.0
+
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package kvstore
+
+import (
+ "bytes"
+ "context"
+ "errors"
+ log "github.com/opencord/voltha-lib-go/v3/pkg/log"
+ "sync"
+ "time"
+ //log "ciena.com/coordinator/common"
+ consulapi "github.com/hashicorp/consul/api"
+)
+
+type channelContextMap struct {
+ ctx context.Context
+ channel chan *Event
+ cancel context.CancelFunc
+}
+
+// ConsulClient represents the consul KV store client
+type ConsulClient struct {
+ session *consulapi.Session
+ sessionID string
+ consul *consulapi.Client
+ doneCh *chan int
+ keyReservations map[string]interface{}
+ watchedChannelsContext map[string][]*channelContextMap
+ writeLock sync.Mutex
+}
+
+// NewConsulClient returns a new client for the Consul KV store
+func NewConsulClient(addr string, timeout int) (*ConsulClient, error) {
+
+ duration := GetDuration(timeout)
+
+ config := consulapi.DefaultConfig()
+ config.Address = addr
+ config.WaitTime = duration
+ consul, err := consulapi.NewClient(config)
+ if err != nil {
+ logger.Error(err)
+ return nil, err
+ }
+
+ doneCh := make(chan int, 1)
+ wChannelsContext := make(map[string][]*channelContextMap)
+ reservations := make(map[string]interface{})
+ return &ConsulClient{consul: consul, doneCh: &doneCh, watchedChannelsContext: wChannelsContext, keyReservations: reservations}, nil
+}
+
+// IsConnectionUp returns whether the connection to the Consul KV store is up
+func (c *ConsulClient) IsConnectionUp(ctx context.Context) bool {
+ logger.Error("Unimplemented function")
+ return false
+}
+
+// List returns an array of key-value pairs with key as a prefix. Timeout defines how long the function will
+// wait for a response
+func (c *ConsulClient) List(ctx context.Context, key string) (map[string]*KVPair, error) {
+
+ deadline, _ := ctx.Deadline()
+ kv := c.consul.KV()
+ var queryOptions consulapi.QueryOptions
+ queryOptions.WaitTime = GetDuration(deadline.Second())
+ // For now we ignore meta data
+ kvps, _, err := kv.List(key, &queryOptions)
+ if err != nil {
+ logger.Error(err)
+ return nil, err
+ }
+ m := make(map[string]*KVPair)
+ for _, kvp := range kvps {
+ m[string(kvp.Key)] = NewKVPair(string(kvp.Key), kvp.Value, string(kvp.Session), 0, -1)
+ }
+ return m, nil
+}
+
+// Get returns a key-value pair for a given key. Timeout defines how long the function will
+// wait for a response
+func (c *ConsulClient) Get(ctx context.Context, key string) (*KVPair, error) {
+
+ deadline, _ := ctx.Deadline()
+ kv := c.consul.KV()
+ var queryOptions consulapi.QueryOptions
+ queryOptions.WaitTime = GetDuration(deadline.Second())
+ // For now we ignore meta data
+ kvp, _, err := kv.Get(key, &queryOptions)
+ if err != nil {
+ logger.Error(err)
+ return nil, err
+ }
+ if kvp != nil {
+ return NewKVPair(string(kvp.Key), kvp.Value, string(kvp.Session), 0, -1), nil
+ }
+
+ return nil, nil
+}
+
+// Put writes a key-value pair to the KV store. Value can only be a string or []byte since the consul API
+// accepts only a []byte as a value for a put operation. Timeout defines how long the function will
+// wait for a response
+func (c *ConsulClient) Put(ctx context.Context, key string, value interface{}) error {
+
+ // Validate that we can create a byte array from the value as consul API expects a byte array
+ var val []byte
+ var er error
+ if val, er = ToByte(value); er != nil {
+ logger.Error(er)
+ return er
+ }
+
+ // Create a key value pair
+ kvp := consulapi.KVPair{Key: key, Value: val}
+ kv := c.consul.KV()
+ var writeOptions consulapi.WriteOptions
+ c.writeLock.Lock()
+ defer c.writeLock.Unlock()
+ _, err := kv.Put(&kvp, &writeOptions)
+ if err != nil {
+ logger.Error(err)
+ return err
+ }
+ return nil
+}
+
+// Delete removes a key from the KV store. Timeout defines how long the function will
+// wait for a response
+func (c *ConsulClient) Delete(ctx context.Context, key string) error {
+ kv := c.consul.KV()
+ var writeOptions consulapi.WriteOptions
+ c.writeLock.Lock()
+ defer c.writeLock.Unlock()
+ _, err := kv.Delete(key, &writeOptions)
+ if err != nil {
+ logger.Error(err)
+ return err
+ }
+ return nil
+}
+
+func (c *ConsulClient) deleteSession() {
+ if c.sessionID != "" {
+ logger.Debug("cleaning-up-session")
+ session := c.consul.Session()
+ _, err := session.Destroy(c.sessionID, nil)
+ if err != nil {
+ logger.Errorw("error-cleaning-session", log.Fields{"session": c.sessionID, "error": err})
+ }
+ }
+ c.sessionID = ""
+ c.session = nil
+}
+
+func (c *ConsulClient) createSession(ttl int64, retries int) (*consulapi.Session, string, error) {
+ session := c.consul.Session()
+ entry := &consulapi.SessionEntry{
+ Behavior: consulapi.SessionBehaviorDelete,
+ TTL: "10s", // strconv.FormatInt(ttl, 10) + "s", // disable ttl
+ }
+
+ for {
+ id, meta, err := session.Create(entry, nil)
+ if err != nil {
+ logger.Errorw("create-session-error", log.Fields{"error": err})
+ if retries == 0 {
+ return nil, "", err
+ }
+ } else if meta.RequestTime == 0 {
+ logger.Errorw("create-session-bad-meta-data", log.Fields{"meta-data": meta})
+ if retries == 0 {
+ return nil, "", errors.New("bad-meta-data")
+ }
+ } else if id == "" {
+ logger.Error("create-session-nil-id")
+ if retries == 0 {
+ return nil, "", errors.New("ID-nil")
+ }
+ } else {
+ return session, id, nil
+ }
+ // If retry param is -1 we will retry indefinitely
+ if retries > 0 {
+ retries--
+ }
+ logger.Debug("retrying-session-create-after-a-second-delay")
+ time.Sleep(time.Duration(1) * time.Second)
+ }
+}
+
+// Helper function to verify mostly whether the content of two interface types are the same. Focus is []byte and
+// string types
+func isEqual(val1 interface{}, val2 interface{}) bool {
+ b1, err := ToByte(val1)
+ b2, er := ToByte(val2)
+ if err == nil && er == nil {
+ return bytes.Equal(b1, b2)
+ }
+ return val1 == val2
+}
+
+// Reserve is invoked to acquire a key and set it to a given value. Value can only be a string or []byte since
+// the consul API accepts only a []byte. Timeout defines how long the function will wait for a response. TTL
+// defines how long that reservation is valid. When TTL expires the key is unreserved by the KV store itself.
+// If the key is acquired then the value returned will be the value passed in. If the key is already acquired
+// then the value assigned to that key will be returned.
+func (c *ConsulClient) Reserve(ctx context.Context, key string, value interface{}, ttl int64) (interface{}, error) {
+
+ // Validate that we can create a byte array from the value as consul API expects a byte array
+ var val []byte
+ var er error
+ if val, er = ToByte(value); er != nil {
+ logger.Error(er)
+ return nil, er
+ }
+
+ // Cleanup any existing session and recreate new ones. A key is reserved against a session
+ if c.sessionID != "" {
+ c.deleteSession()
+ }
+
+ // Clear session if reservation is not successful
+ reservationSuccessful := false
+ defer func() {
+ if !reservationSuccessful {
+ logger.Debug("deleting-session")
+ c.deleteSession()
+ }
+ }()
+
+ session, sessionID, err := c.createSession(ttl, -1)
+ if err != nil {
+ logger.Errorw("no-session-created", log.Fields{"error": err})
+ return "", errors.New("no-session-created")
+ }
+ logger.Debugw("session-created", log.Fields{"session-id": sessionID})
+ c.sessionID = sessionID
+ c.session = session
+
+ // Try to grap the Key using the session
+ kv := c.consul.KV()
+ kvp := consulapi.KVPair{Key: key, Value: val, Session: c.sessionID}
+ result, _, err := kv.Acquire(&kvp, nil)
+ if err != nil {
+ logger.Errorw("error-acquiring-keys", log.Fields{"error": err})
+ return nil, err
+ }
+
+ logger.Debugw("key-acquired", log.Fields{"key": key, "status": result})
+
+ // Irrespective whether we were successful in acquiring the key, let's read it back and see if it's us.
+ m, err := c.Get(ctx, key)
+ if err != nil {
+ return nil, err
+ }
+ if m != nil {
+ logger.Debugw("response-received", log.Fields{"key": m.Key, "m.value": string(m.Value.([]byte)), "value": value})
+ if m.Key == key && isEqual(m.Value, value) {
+ // My reservation is successful - register it. For now, support is only for 1 reservation per key
+ // per session.
+ reservationSuccessful = true
+ c.writeLock.Lock()
+ c.keyReservations[key] = m.Value
+ c.writeLock.Unlock()
+ return m.Value, nil
+ }
+ // My reservation has failed. Return the owner of that key
+ return m.Value, nil
+ }
+ return nil, nil
+}
+
+// ReleaseAllReservations releases all key reservations previously made (using Reserve API)
+func (c *ConsulClient) ReleaseAllReservations(ctx context.Context) error {
+ kv := c.consul.KV()
+ var kvp consulapi.KVPair
+ var result bool
+ var err error
+
+ c.writeLock.Lock()
+ defer c.writeLock.Unlock()
+
+ for key, value := range c.keyReservations {
+ kvp = consulapi.KVPair{Key: key, Value: value.([]byte), Session: c.sessionID}
+ result, _, err = kv.Release(&kvp, nil)
+ if err != nil {
+ logger.Errorw("cannot-release-reservation", log.Fields{"key": key, "error": err})
+ return err
+ }
+ if !result {
+ logger.Errorw("cannot-release-reservation", log.Fields{"key": key})
+ }
+ delete(c.keyReservations, key)
+ }
+ return nil
+}
+
+// ReleaseReservation releases reservation for a specific key.
+func (c *ConsulClient) ReleaseReservation(ctx context.Context, key string) error {
+ var ok bool
+ var reservedValue interface{}
+ c.writeLock.Lock()
+ defer c.writeLock.Unlock()
+ if reservedValue, ok = c.keyReservations[key]; !ok {
+ return errors.New("key-not-reserved:" + key)
+ }
+ // Release the reservation
+ kv := c.consul.KV()
+ kvp := consulapi.KVPair{Key: key, Value: reservedValue.([]byte), Session: c.sessionID}
+
+ result, _, er := kv.Release(&kvp, nil)
+ if er != nil {
+ return er
+ }
+ // Remove that key entry on success
+ if result {
+ delete(c.keyReservations, key)
+ return nil
+ }
+ return errors.New("key-cannot-be-unreserved")
+}
+
+// RenewReservation renews a reservation. A reservation will go stale after the specified TTL (Time To Live)
+// period specified when reserving the key
+func (c *ConsulClient) RenewReservation(ctx context.Context, key string) error {
+ // In the case of Consul, renew reservation of a reserve key only require renewing the client session.
+
+ c.writeLock.Lock()
+ defer c.writeLock.Unlock()
+
+ // Verify the key was reserved
+ if _, ok := c.keyReservations[key]; !ok {
+ return errors.New("key-not-reserved")
+ }
+
+ if c.session == nil {
+ return errors.New("no-session-exist")
+ }
+
+ var writeOptions consulapi.WriteOptions
+ if _, _, err := c.session.Renew(c.sessionID, &writeOptions); err != nil {
+ return err
+ }
+ return nil
+}
+
+// Watch provides the watch capability on a given key. It returns a channel onto which the callee needs to
+// listen to receive Events.
+func (c *ConsulClient) Watch(ctx context.Context, key string, withPrefix bool) chan *Event {
+
+ // Create a new channel
+ ch := make(chan *Event, maxClientChannelBufferSize)
+
+ // Create a context to track this request
+ watchContext, cFunc := context.WithCancel(context.Background())
+
+ // Save the channel and context reference for later
+ c.writeLock.Lock()
+ defer c.writeLock.Unlock()
+ ccm := channelContextMap{channel: ch, ctx: watchContext, cancel: cFunc}
+ c.watchedChannelsContext[key] = append(c.watchedChannelsContext[key], &ccm)
+
+ // Launch a go routine to listen for updates
+ go c.listenForKeyChange(watchContext, key, ch)
+
+ return ch
+}
+
+// CloseWatch closes a specific watch. Both the key and the channel are required when closing a watch as there
+// may be multiple listeners on the same key. The previously created channel serves as a key
+func (c *ConsulClient) CloseWatch(key string, ch chan *Event) {
+ // First close the context
+ var ok bool
+ var watchedChannelsContexts []*channelContextMap
+ c.writeLock.Lock()
+ defer c.writeLock.Unlock()
+ if watchedChannelsContexts, ok = c.watchedChannelsContext[key]; !ok {
+ logger.Errorw("key-has-no-watched-context-or-channel", log.Fields{"key": key})
+ return
+ }
+ // Look for the channels
+ var pos = -1
+ for i, chCtxMap := range watchedChannelsContexts {
+ if chCtxMap.channel == ch {
+ logger.Debug("channel-found")
+ chCtxMap.cancel()
+ //close the channel
+ close(ch)
+ pos = i
+ break
+ }
+ }
+ // Remove that entry if present
+ if pos >= 0 {
+ c.watchedChannelsContext[key] = append(c.watchedChannelsContext[key][:pos], c.watchedChannelsContext[key][pos+1:]...)
+ }
+ logger.Debugw("watched-channel-exiting", log.Fields{"key": key, "channel": c.watchedChannelsContext[key]})
+}
+
+func (c *ConsulClient) isKVEqual(kv1 *consulapi.KVPair, kv2 *consulapi.KVPair) bool {
+ if (kv1 == nil) && (kv2 == nil) {
+ return true
+ } else if (kv1 == nil) || (kv2 == nil) {
+ return false
+ }
+ // Both the KV should be non-null here
+ if kv1.Key != kv2.Key ||
+ !bytes.Equal(kv1.Value, kv2.Value) ||
+ kv1.Session != kv2.Session ||
+ kv1.LockIndex != kv2.LockIndex ||
+ kv1.ModifyIndex != kv2.ModifyIndex {
+ return false
+ }
+ return true
+}
+
+func (c *ConsulClient) listenForKeyChange(watchContext context.Context, key string, ch chan *Event) {
+ logger.Debugw("start-watching-channel", log.Fields{"key": key, "channel": ch})
+
+ defer c.CloseWatch(key, ch)
+ duration := GetDuration(defaultKVGetTimeout)
+ kv := c.consul.KV()
+ var queryOptions consulapi.QueryOptions
+ queryOptions.WaitTime = duration
+
+ // Get the existing value, if any
+ previousKVPair, meta, err := kv.Get(key, &queryOptions)
+ if err != nil {
+ logger.Debug(err)
+ }
+ lastIndex := meta.LastIndex
+
+ // Wait for change. Push any change onto the channel and keep waiting for new update
+ //var waitOptions consulapi.QueryOptions
+ var pair *consulapi.KVPair
+ //watchContext, _ := context.WithCancel(context.Background())
+ waitOptions := queryOptions.WithContext(watchContext)
+ for {
+ //waitOptions = consulapi.QueryOptions{WaitIndex: lastIndex}
+ waitOptions.WaitIndex = lastIndex
+ pair, meta, err = kv.Get(key, waitOptions)
+ select {
+ case <-watchContext.Done():
+ logger.Debug("done-event-received-exiting")
+ return
+ default:
+ if err != nil {
+ logger.Warnw("error-from-watch", log.Fields{"error": err})
+ ch <- NewEvent(CONNECTIONDOWN, key, []byte(""), -1)
+ } else {
+ logger.Debugw("index-state", log.Fields{"lastindex": lastIndex, "newindex": meta.LastIndex, "key": key})
+ }
+ }
+ if err != nil {
+ logger.Debug(err)
+ // On error, block for 10 milliseconds to prevent endless loop
+ time.Sleep(10 * time.Millisecond)
+ } else if meta.LastIndex <= lastIndex {
+ logger.Info("no-index-change-or-negative")
+ } else {
+ logger.Debugw("update-received", log.Fields{"pair": pair})
+ if pair == nil {
+ ch <- NewEvent(DELETE, key, []byte(""), -1)
+ } else if !c.isKVEqual(pair, previousKVPair) {
+ // Push the change onto the channel if the data has changed
+ // For now just assume it's a PUT change
+ logger.Debugw("pair-details", log.Fields{"session": pair.Session, "key": pair.Key, "value": pair.Value})
+ ch <- NewEvent(PUT, pair.Key, pair.Value, -1)
+ }
+ previousKVPair = pair
+ lastIndex = meta.LastIndex
+ }
+ }
+}
+
+// Close closes the KV store client
+func (c *ConsulClient) Close() {
+ var writeOptions consulapi.WriteOptions
+ // Inform any goroutine it's time to say goodbye.
+ c.writeLock.Lock()
+ defer c.writeLock.Unlock()
+ if c.doneCh != nil {
+ close(*c.doneCh)
+ }
+
+ // Clear the sessionID
+ if _, err := c.consul.Session().Destroy(c.sessionID, &writeOptions); err != nil {
+ logger.Errorw("error-closing-client", log.Fields{"error": err})
+ }
+}
+
+func (c *ConsulClient) AcquireLock(ctx context.Context, lockName string, timeout int) error {
+ return nil
+}
+
+func (c *ConsulClient) ReleaseLock(lockName string) error {
+ return nil
+}
diff --git a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore/etcdclient.go b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore/etcdclient.go
new file mode 100644
index 0000000..d38f0f6
--- /dev/null
+++ b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore/etcdclient.go
@@ -0,0 +1,485 @@
+/*
+ * Copyright 2018-present Open Networking Foundation
+
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+
+ * http://www.apache.org/licenses/LICENSE-2.0
+
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package kvstore
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "sync"
+
+ "github.com/opencord/voltha-lib-go/v3/pkg/log"
+ v3Client "go.etcd.io/etcd/clientv3"
+ v3Concurrency "go.etcd.io/etcd/clientv3/concurrency"
+ v3rpcTypes "go.etcd.io/etcd/etcdserver/api/v3rpc/rpctypes"
+)
+
+// EtcdClient represents the Etcd KV store client
+type EtcdClient struct {
+ ectdAPI *v3Client.Client
+ keyReservations map[string]*v3Client.LeaseID
+ watchedChannels sync.Map
+ keyReservationsLock sync.RWMutex
+ lockToMutexMap map[string]*v3Concurrency.Mutex
+ lockToSessionMap map[string]*v3Concurrency.Session
+ lockToMutexLock sync.Mutex
+}
+
+// NewEtcdClient returns a new client for the Etcd KV store
+func NewEtcdClient(addr string, timeout int) (*EtcdClient, error) {
+ duration := GetDuration(timeout)
+
+ c, err := v3Client.New(v3Client.Config{
+ Endpoints: []string{addr},
+ DialTimeout: duration,
+ })
+ if err != nil {
+ logger.Error(err)
+ return nil, err
+ }
+
+ reservations := make(map[string]*v3Client.LeaseID)
+ lockMutexMap := make(map[string]*v3Concurrency.Mutex)
+ lockSessionMap := make(map[string]*v3Concurrency.Session)
+
+ return &EtcdClient{ectdAPI: c, keyReservations: reservations, lockToMutexMap: lockMutexMap,
+ lockToSessionMap: lockSessionMap}, nil
+}
+
+// IsConnectionUp returns whether the connection to the Etcd KV store is up. If a timeout occurs then
+// it is assumed the connection is down or unreachable.
+func (c *EtcdClient) IsConnectionUp(ctx context.Context) bool {
+ // Let's try to get a non existent key. If the connection is up then there will be no error returned.
+ if _, err := c.Get(ctx, "non-existent-key"); err != nil {
+ return false
+ }
+ //cancel()
+ return true
+}
+
+// List returns an array of key-value pairs with key as a prefix. Timeout defines how long the function will
+// wait for a response
+func (c *EtcdClient) List(ctx context.Context, key string) (map[string]*KVPair, error) {
+ resp, err := c.ectdAPI.Get(ctx, key, v3Client.WithPrefix())
+ if err != nil {
+ logger.Error(err)
+ return nil, err
+ }
+ m := make(map[string]*KVPair)
+ for _, ev := range resp.Kvs {
+ m[string(ev.Key)] = NewKVPair(string(ev.Key), ev.Value, "", ev.Lease, ev.Version)
+ }
+ return m, nil
+}
+
+// Get returns a key-value pair for a given key. Timeout defines how long the function will
+// wait for a response
+func (c *EtcdClient) Get(ctx context.Context, key string) (*KVPair, error) {
+
+ resp, err := c.ectdAPI.Get(ctx, key)
+
+ if err != nil {
+ logger.Error(err)
+ return nil, err
+ }
+ for _, ev := range resp.Kvs {
+ // Only one value is returned
+ return NewKVPair(string(ev.Key), ev.Value, "", ev.Lease, ev.Version), nil
+ }
+ return nil, nil
+}
+
+// Put writes a key-value pair to the KV store. Value can only be a string or []byte since the etcd API
+// accepts only a string as a value for a put operation. Timeout defines how long the function will
+// wait for a response
+func (c *EtcdClient) Put(ctx context.Context, key string, value interface{}) error {
+
+ // Validate that we can convert value to a string as etcd API expects a string
+ var val string
+ var er error
+ if val, er = ToString(value); er != nil {
+ return fmt.Errorf("unexpected-type-%T", value)
+ }
+
+ var err error
+ // Check if there is already a lease for this key - if there is then use it, otherwise a PUT will make
+ // that KV key permanent instead of automatically removing it after a lease expiration
+ c.keyReservationsLock.RLock()
+ leaseID, ok := c.keyReservations[key]
+ c.keyReservationsLock.RUnlock()
+ if ok {
+ _, err = c.ectdAPI.Put(ctx, key, val, v3Client.WithLease(*leaseID))
+ } else {
+ _, err = c.ectdAPI.Put(ctx, key, val)
+ }
+
+ if err != nil {
+ switch err {
+ case context.Canceled:
+ logger.Warnw("context-cancelled", log.Fields{"error": err})
+ case context.DeadlineExceeded:
+ logger.Warnw("context-deadline-exceeded", log.Fields{"error": err})
+ case v3rpcTypes.ErrEmptyKey:
+ logger.Warnw("etcd-client-error", log.Fields{"error": err})
+ default:
+ logger.Warnw("bad-endpoints", log.Fields{"error": err})
+ }
+ return err
+ }
+ return nil
+}
+
+// Delete removes a key from the KV store. Timeout defines how long the function will
+// wait for a response
+func (c *EtcdClient) Delete(ctx context.Context, key string) error {
+
+ // delete the key
+ if _, err := c.ectdAPI.Delete(ctx, key); err != nil {
+ logger.Errorw("failed-to-delete-key", log.Fields{"key": key, "error": err})
+ return err
+ }
+ logger.Debugw("key(s)-deleted", log.Fields{"key": key})
+ return nil
+}
+
+// Reserve is invoked to acquire a key and set it to a given value. Value can only be a string or []byte since
+// the etcd API accepts only a string. Timeout defines how long the function will wait for a response. TTL
+// defines how long that reservation is valid. When TTL expires the key is unreserved by the KV store itself.
+// If the key is acquired then the value returned will be the value passed in. If the key is already acquired
+// then the value assigned to that key will be returned.
+func (c *EtcdClient) Reserve(ctx context.Context, key string, value interface{}, ttl int64) (interface{}, error) {
+ // Validate that we can convert value to a string as etcd API expects a string
+ var val string
+ var er error
+ if val, er = ToString(value); er != nil {
+ return nil, fmt.Errorf("unexpected-type%T", value)
+ }
+
+ resp, err := c.ectdAPI.Grant(ctx, ttl)
+ if err != nil {
+ logger.Error(err)
+ return nil, err
+ }
+ // Register the lease id
+ c.keyReservationsLock.Lock()
+ c.keyReservations[key] = &resp.ID
+ c.keyReservationsLock.Unlock()
+
+ // Revoke lease if reservation is not successful
+ reservationSuccessful := false
+ defer func() {
+ if !reservationSuccessful {
+ if err = c.ReleaseReservation(context.Background(), key); err != nil {
+ logger.Error("cannot-release-lease")
+ }
+ }
+ }()
+
+ // Try to grap the Key with the above lease
+ c.ectdAPI.Txn(context.Background())
+ txn := c.ectdAPI.Txn(context.Background())
+ txn = txn.If(v3Client.Compare(v3Client.Version(key), "=", 0))
+ txn = txn.Then(v3Client.OpPut(key, val, v3Client.WithLease(resp.ID)))
+ txn = txn.Else(v3Client.OpGet(key))
+ result, er := txn.Commit()
+ if er != nil {
+ return nil, er
+ }
+
+ if !result.Succeeded {
+ // Verify whether we are already the owner of that Key
+ if len(result.Responses) > 0 &&
+ len(result.Responses[0].GetResponseRange().Kvs) > 0 {
+ kv := result.Responses[0].GetResponseRange().Kvs[0]
+ if string(kv.Value) == val {
+ reservationSuccessful = true
+ return value, nil
+ }
+ return kv.Value, nil
+ }
+ } else {
+ // Read the Key to ensure this is our Key
+ m, err := c.Get(ctx, key)
+ if err != nil {
+ return nil, err
+ }
+ if m != nil {
+ if m.Key == key && isEqual(m.Value, value) {
+ // My reservation is successful - register it. For now, support is only for 1 reservation per key
+ // per session.
+ reservationSuccessful = true
+ return value, nil
+ }
+ // My reservation has failed. Return the owner of that key
+ return m.Value, nil
+ }
+ }
+ return nil, nil
+}
+
+// ReleaseAllReservations releases all key reservations previously made (using Reserve API)
+func (c *EtcdClient) ReleaseAllReservations(ctx context.Context) error {
+ c.keyReservationsLock.Lock()
+ defer c.keyReservationsLock.Unlock()
+
+ for key, leaseID := range c.keyReservations {
+ _, err := c.ectdAPI.Revoke(ctx, *leaseID)
+ if err != nil {
+ logger.Errorw("cannot-release-reservation", log.Fields{"key": key, "error": err})
+ return err
+ }
+ delete(c.keyReservations, key)
+ }
+ return nil
+}
+
+// ReleaseReservation releases reservation for a specific key.
+func (c *EtcdClient) ReleaseReservation(ctx context.Context, key string) error {
+ // Get the leaseid using the key
+ logger.Debugw("Release-reservation", log.Fields{"key": key})
+ var ok bool
+ var leaseID *v3Client.LeaseID
+ c.keyReservationsLock.Lock()
+ defer c.keyReservationsLock.Unlock()
+ if leaseID, ok = c.keyReservations[key]; !ok {
+ return nil
+ }
+
+ if leaseID != nil {
+ _, err := c.ectdAPI.Revoke(ctx, *leaseID)
+ if err != nil {
+ logger.Error(err)
+ return err
+ }
+ delete(c.keyReservations, key)
+ }
+ return nil
+}
+
+// RenewReservation renews a reservation. A reservation will go stale after the specified TTL (Time To Live)
+// period specified when reserving the key
+func (c *EtcdClient) RenewReservation(ctx context.Context, key string) error {
+ // Get the leaseid using the key
+ var ok bool
+ var leaseID *v3Client.LeaseID
+ c.keyReservationsLock.RLock()
+ leaseID, ok = c.keyReservations[key]
+ c.keyReservationsLock.RUnlock()
+
+ if !ok {
+ return errors.New("key-not-reserved")
+ }
+
+ if leaseID != nil {
+ _, err := c.ectdAPI.KeepAliveOnce(ctx, *leaseID)
+ if err != nil {
+ logger.Errorw("lease-may-have-expired", log.Fields{"error": err})
+ return err
+ }
+ } else {
+ return errors.New("lease-expired")
+ }
+ return nil
+}
+
+// Watch provides the watch capability on a given key. It returns a channel onto which the callee needs to
+// listen to receive Events.
+func (c *EtcdClient) Watch(ctx context.Context, key string, withPrefix bool) chan *Event {
+ w := v3Client.NewWatcher(c.ectdAPI)
+ ctx, cancel := context.WithCancel(ctx)
+ var channel v3Client.WatchChan
+ if withPrefix {
+ channel = w.Watch(ctx, key, v3Client.WithPrefix())
+ } else {
+ channel = w.Watch(ctx, key)
+ }
+
+ // Create a new channel
+ ch := make(chan *Event, maxClientChannelBufferSize)
+
+ // Keep track of the created channels so they can be closed when required
+ channelMap := make(map[chan *Event]v3Client.Watcher)
+ channelMap[ch] = w
+
+ channelMaps := c.addChannelMap(key, channelMap)
+
+ // Changing the log field (from channelMaps) as the underlying logger cannot format the map of channels into a
+ // json format.
+ logger.Debugw("watched-channels", log.Fields{"len": len(channelMaps)})
+ // Launch a go routine to listen for updates
+ go c.listenForKeyChange(channel, ch, cancel)
+
+ return ch
+
+}
+
+func (c *EtcdClient) addChannelMap(key string, channelMap map[chan *Event]v3Client.Watcher) []map[chan *Event]v3Client.Watcher {
+ var channels interface{}
+ var exists bool
+
+ if channels, exists = c.watchedChannels.Load(key); exists {
+ channels = append(channels.([]map[chan *Event]v3Client.Watcher), channelMap)
+ } else {
+ channels = []map[chan *Event]v3Client.Watcher{channelMap}
+ }
+ c.watchedChannels.Store(key, channels)
+
+ return channels.([]map[chan *Event]v3Client.Watcher)
+}
+
+func (c *EtcdClient) removeChannelMap(key string, pos int) []map[chan *Event]v3Client.Watcher {
+ var channels interface{}
+ var exists bool
+
+ if channels, exists = c.watchedChannels.Load(key); exists {
+ channels = append(channels.([]map[chan *Event]v3Client.Watcher)[:pos], channels.([]map[chan *Event]v3Client.Watcher)[pos+1:]...)
+ c.watchedChannels.Store(key, channels)
+ }
+
+ return channels.([]map[chan *Event]v3Client.Watcher)
+}
+
+func (c *EtcdClient) getChannelMaps(key string) ([]map[chan *Event]v3Client.Watcher, bool) {
+ var channels interface{}
+ var exists bool
+
+ channels, exists = c.watchedChannels.Load(key)
+
+ if channels == nil {
+ return nil, exists
+ }
+
+ return channels.([]map[chan *Event]v3Client.Watcher), exists
+}
+
+// CloseWatch closes a specific watch. Both the key and the channel are required when closing a watch as there
+// may be multiple listeners on the same key. The previously created channel serves as a key
+func (c *EtcdClient) CloseWatch(key string, ch chan *Event) {
+ // Get the array of channels mapping
+ var watchedChannels []map[chan *Event]v3Client.Watcher
+ var ok bool
+
+ if watchedChannels, ok = c.getChannelMaps(key); !ok {
+ logger.Warnw("key-has-no-watched-channels", log.Fields{"key": key})
+ return
+ }
+ // Look for the channels
+ var pos = -1
+ for i, chMap := range watchedChannels {
+ if t, ok := chMap[ch]; ok {
+ logger.Debug("channel-found")
+ // Close the etcd watcher before the client channel. This should close the etcd channel as well
+ if err := t.Close(); err != nil {
+ logger.Errorw("watcher-cannot-be-closed", log.Fields{"key": key, "error": err})
+ }
+ pos = i
+ break
+ }
+ }
+
+ channelMaps, _ := c.getChannelMaps(key)
+ // Remove that entry if present
+ if pos >= 0 {
+ channelMaps = c.removeChannelMap(key, pos)
+ }
+ logger.Infow("watcher-channel-exiting", log.Fields{"key": key, "channel": channelMaps})
+}
+
+func (c *EtcdClient) listenForKeyChange(channel v3Client.WatchChan, ch chan<- *Event, cancel context.CancelFunc) {
+ logger.Debug("start-listening-on-channel ...")
+ defer cancel()
+ defer close(ch)
+ for resp := range channel {
+ for _, ev := range resp.Events {
+ ch <- NewEvent(getEventType(ev), ev.Kv.Key, ev.Kv.Value, ev.Kv.Version)
+ }
+ }
+ logger.Debug("stop-listening-on-channel ...")
+}
+
+func getEventType(event *v3Client.Event) int {
+ switch event.Type {
+ case v3Client.EventTypePut:
+ return PUT
+ case v3Client.EventTypeDelete:
+ return DELETE
+ }
+ return UNKNOWN
+}
+
+// Close closes the KV store client
+func (c *EtcdClient) Close() {
+ if err := c.ectdAPI.Close(); err != nil {
+ logger.Errorw("error-closing-client", log.Fields{"error": err})
+ }
+}
+
+func (c *EtcdClient) addLockName(lockName string, lock *v3Concurrency.Mutex, session *v3Concurrency.Session) {
+ c.lockToMutexLock.Lock()
+ defer c.lockToMutexLock.Unlock()
+ c.lockToMutexMap[lockName] = lock
+ c.lockToSessionMap[lockName] = session
+}
+
+func (c *EtcdClient) deleteLockName(lockName string) {
+ c.lockToMutexLock.Lock()
+ defer c.lockToMutexLock.Unlock()
+ delete(c.lockToMutexMap, lockName)
+ delete(c.lockToSessionMap, lockName)
+}
+
+func (c *EtcdClient) getLock(lockName string) (*v3Concurrency.Mutex, *v3Concurrency.Session) {
+ c.lockToMutexLock.Lock()
+ defer c.lockToMutexLock.Unlock()
+ var lock *v3Concurrency.Mutex
+ var session *v3Concurrency.Session
+ if l, exist := c.lockToMutexMap[lockName]; exist {
+ lock = l
+ }
+ if s, exist := c.lockToSessionMap[lockName]; exist {
+ session = s
+ }
+ return lock, session
+}
+
+func (c *EtcdClient) AcquireLock(ctx context.Context, lockName string, timeout int) error {
+ session, _ := v3Concurrency.NewSession(c.ectdAPI, v3Concurrency.WithContext(ctx))
+ mu := v3Concurrency.NewMutex(session, "/devicelock_"+lockName)
+ if err := mu.Lock(context.Background()); err != nil {
+ //cancel()
+ return err
+ }
+ c.addLockName(lockName, mu, session)
+ return nil
+}
+
+func (c *EtcdClient) ReleaseLock(lockName string) error {
+ lock, session := c.getLock(lockName)
+ var err error
+ if lock != nil {
+ if e := lock.Unlock(context.Background()); e != nil {
+ err = e
+ }
+ }
+ if session != nil {
+ if e := session.Close(); e != nil {
+ err = e
+ }
+ }
+ c.deleteLockName(lockName)
+
+ return err
+}
diff --git a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore/kvutils.go b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore/kvutils.go
new file mode 100644
index 0000000..cf9a95c
--- /dev/null
+++ b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore/kvutils.go
@@ -0,0 +1,56 @@
+/*
+ * Copyright 2018-present Open Networking Foundation
+
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+
+ * http://www.apache.org/licenses/LICENSE-2.0
+
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package kvstore
+
+import (
+ "fmt"
+ "time"
+)
+
+// GetDuration converts a timeout value from int to duration. If the timeout value is
+// either not set of -ve then we default KV timeout (configurable) is used.
+func GetDuration(timeout int) time.Duration {
+ if timeout <= 0 {
+ return defaultKVGetTimeout * time.Second
+ }
+ return time.Duration(timeout) * time.Second
+}
+
+// ToString converts an interface value to a string. The interface should either be of
+// a string type or []byte. Otherwise, an error is returned.
+func ToString(value interface{}) (string, error) {
+ switch t := value.(type) {
+ case []byte:
+ return string(value.([]byte)), nil
+ case string:
+ return value.(string), nil
+ default:
+ return "", fmt.Errorf("unexpected-type-%T", t)
+ }
+}
+
+// ToByte converts an interface value to a []byte. The interface should either be of
+// a string type or []byte. Otherwise, an error is returned.
+func ToByte(value interface{}) ([]byte, error) {
+ switch t := value.(type) {
+ case []byte:
+ return value.([]byte), nil
+ case string:
+ return []byte(value.(string)), nil
+ default:
+ return nil, fmt.Errorf("unexpected-type-%T", t)
+ }
+}
diff --git a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/log/log.go b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/log/log.go
new file mode 100644
index 0000000..47fa3fb
--- /dev/null
+++ b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/log/log.go
@@ -0,0 +1,798 @@
+/*
+ * Copyright 2018-present Open Networking Foundation
+
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+
+ * http://www.apache.org/licenses/LICENSE-2.0
+
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+//Package log provides a structured Logger interface implemented using zap logger. It provides the following capabilities:
+//1. Package level logging - a go package can register itself (AddPackage) and have a logger created for that package.
+//2. Dynamic log level change - for all registered packages (SetAllLogLevel)
+//3. Dynamic log level change - for a given package (SetPackageLogLevel)
+//4. Provides a default logger for unregistered packages
+//5. Allow key-value pairs to be added to a logger(UpdateLogger) or all loggers (UpdateAllLoggers) at run time
+//6. Add to the log output the location where the log was invoked (filename.functionname.linenumber)
+//
+// Using package-level logging (recommended approach). In the examples below, log refers to this log package.
+// 1. In the appropriate package add the following in the init section of the package. The log level can be changed
+// and any number of default fields can be added as well. The log level specifies the lowest log level that will be
+// in the output while the fields will be automatically added to all log printouts.
+//
+// log.AddPackage(mylog.JSON, log.WarnLevel, log.Fields{"anyFieldName": "any value"})
+//
+//2. In the calling package, just invoke any of the publicly available functions of the logger. Here is an example
+// to write an Info log with additional fields:
+//
+//log.Infow("An example", mylog.Fields{"myStringOutput": "output", "myIntOutput": 2})
+//
+//3. To dynamically change the log level, you can use 1)SetLogLevel from inside your package or 2) SetPackageLogLevel
+// from anywhere or 3) SetAllLogLevel from anywhere.
+//
+
+package log
+
+import (
+ "errors"
+ "fmt"
+ zp "go.uber.org/zap"
+ zc "go.uber.org/zap/zapcore"
+ "path"
+ "runtime"
+ "strings"
+)
+
+type LogLevel int8
+
+const (
+ // DebugLevel logs a message at debug level
+ DebugLevel = LogLevel(iota)
+ // InfoLevel logs a message at info level
+ InfoLevel
+ // WarnLevel logs a message at warning level
+ WarnLevel
+ // ErrorLevel logs a message at error level
+ ErrorLevel
+ // FatalLevel logs a message, then calls os.Exit(1).
+ FatalLevel
+)
+
+// CONSOLE formats the log for the console, mostly used during development
+const CONSOLE = "console"
+
+// JSON formats the log using json format, mostly used by an automated logging system consumption
+const JSON = "json"
+
+// Logger represents an abstract logging interface. Any logging implementation used
+// will need to abide by this interface
+type Logger interface {
+ Debug(...interface{})
+ Debugln(...interface{})
+ Debugf(string, ...interface{})
+ Debugw(string, Fields)
+
+ Info(...interface{})
+ Infoln(...interface{})
+ Infof(string, ...interface{})
+ Infow(string, Fields)
+
+ Warn(...interface{})
+ Warnln(...interface{})
+ Warnf(string, ...interface{})
+ Warnw(string, Fields)
+
+ Error(...interface{})
+ Errorln(...interface{})
+ Errorf(string, ...interface{})
+ Errorw(string, Fields)
+
+ Fatal(...interface{})
+ Fatalln(...interface{})
+ Fatalf(string, ...interface{})
+ Fatalw(string, Fields)
+
+ With(Fields) Logger
+
+ // The following are added to be able to use this logger as a gRPC LoggerV2 if needed
+ //
+ Warning(...interface{})
+ Warningln(...interface{})
+ Warningf(string, ...interface{})
+
+ // V reports whether verbosity level l is at least the requested verbose level.
+ V(l LogLevel) bool
+
+ //Returns the log level of this specific logger
+ GetLogLevel() LogLevel
+}
+
+// Fields is used as key-value pairs for structured logging
+type Fields map[string]interface{}
+
+var defaultLogger *logger
+var cfg zp.Config
+
+var loggers map[string]*logger
+var cfgs map[string]zp.Config
+
+type logger struct {
+ log *zp.SugaredLogger
+ parent *zp.Logger
+ packageName string
+}
+
+func logLevelToAtomicLevel(l LogLevel) zp.AtomicLevel {
+ switch l {
+ case DebugLevel:
+ return zp.NewAtomicLevelAt(zc.DebugLevel)
+ case InfoLevel:
+ return zp.NewAtomicLevelAt(zc.InfoLevel)
+ case WarnLevel:
+ return zp.NewAtomicLevelAt(zc.WarnLevel)
+ case ErrorLevel:
+ return zp.NewAtomicLevelAt(zc.ErrorLevel)
+ case FatalLevel:
+ return zp.NewAtomicLevelAt(zc.FatalLevel)
+ }
+ return zp.NewAtomicLevelAt(zc.ErrorLevel)
+}
+
+func logLevelToLevel(l LogLevel) zc.Level {
+ switch l {
+ case DebugLevel:
+ return zc.DebugLevel
+ case InfoLevel:
+ return zc.InfoLevel
+ case WarnLevel:
+ return zc.WarnLevel
+ case ErrorLevel:
+ return zc.ErrorLevel
+ case FatalLevel:
+ return zc.FatalLevel
+ }
+ return zc.ErrorLevel
+}
+
+func levelToLogLevel(l zc.Level) LogLevel {
+ switch l {
+ case zc.DebugLevel:
+ return DebugLevel
+ case zc.InfoLevel:
+ return InfoLevel
+ case zc.WarnLevel:
+ return WarnLevel
+ case zc.ErrorLevel:
+ return ErrorLevel
+ case zc.FatalLevel:
+ return FatalLevel
+ }
+ return ErrorLevel
+}
+
+func StringToLogLevel(l string) (LogLevel, error) {
+ switch strings.ToUpper(l) {
+ case "DEBUG":
+ return DebugLevel, nil
+ case "INFO":
+ return InfoLevel, nil
+ case "WARN":
+ return WarnLevel, nil
+ case "ERROR":
+ return ErrorLevel, nil
+ case "FATAL":
+ return FatalLevel, nil
+ }
+ return 0, errors.New("Given LogLevel is invalid : " + l)
+}
+
+func LogLevelToString(l LogLevel) (string, error) {
+ switch l {
+ case DebugLevel:
+ return "DEBUG", nil
+ case InfoLevel:
+ return "INFO", nil
+ case WarnLevel:
+ return "WARN", nil
+ case ErrorLevel:
+ return "ERROR", nil
+ case FatalLevel:
+ return "FATAL", nil
+ }
+ return "", errors.New("Given LogLevel is invalid " + string(l))
+}
+
+func getDefaultConfig(outputType string, level LogLevel, defaultFields Fields) zp.Config {
+ return zp.Config{
+ Level: logLevelToAtomicLevel(level),
+ Encoding: outputType,
+ Development: true,
+ OutputPaths: []string{"stdout"},
+ ErrorOutputPaths: []string{"stderr"},
+ InitialFields: defaultFields,
+ EncoderConfig: zc.EncoderConfig{
+ LevelKey: "level",
+ MessageKey: "msg",
+ TimeKey: "ts",
+ CallerKey: "caller",
+ StacktraceKey: "stacktrace",
+ LineEnding: zc.DefaultLineEnding,
+ EncodeLevel: zc.LowercaseLevelEncoder,
+ EncodeTime: zc.ISO8601TimeEncoder,
+ EncodeDuration: zc.SecondsDurationEncoder,
+ EncodeCaller: zc.ShortCallerEncoder,
+ },
+ }
+}
+
+// SetLogger needs to be invoked before the logger API can be invoked. This function
+// initialize the default logger (zap's sugaredlogger)
+func SetDefaultLogger(outputType string, level LogLevel, defaultFields Fields) (Logger, error) {
+ // Build a custom config using zap
+ cfg = getDefaultConfig(outputType, level, defaultFields)
+
+ l, err := cfg.Build(zp.AddCallerSkip(1))
+ if err != nil {
+ return nil, err
+ }
+
+ defaultLogger = &logger{
+ log: l.Sugar(),
+ parent: l,
+ }
+
+ return defaultLogger, nil
+}
+
+// AddPackage registers a package to the log map. Each package gets its own logger which allows
+// its config (loglevel) to be changed dynamically without interacting with the other packages.
+// outputType is JSON, level is the lowest level log to output with this logger and defaultFields is a map of
+// key-value pairs to always add to the output.
+// Note: AddPackage also returns a reference to the actual logger. If a calling package uses this reference directly
+//instead of using the publicly available functions in this log package then a number of functionalities will not
+// be available to it, notably log tracing with filename.functionname.linenumber annotation.
+//
+// pkgNames parameter should be used for testing only as this function detects the caller's package.
+func AddPackage(outputType string, level LogLevel, defaultFields Fields, pkgNames ...string) (Logger, error) {
+ if cfgs == nil {
+ cfgs = make(map[string]zp.Config)
+ }
+ if loggers == nil {
+ loggers = make(map[string]*logger)
+ }
+
+ var pkgName string
+ for _, name := range pkgNames {
+ pkgName = name
+ break
+ }
+ if pkgName == "" {
+ pkgName, _, _, _ = getCallerInfo()
+ }
+
+ if _, exist := loggers[pkgName]; exist {
+ return loggers[pkgName], nil
+ }
+
+ cfgs[pkgName] = getDefaultConfig(outputType, level, defaultFields)
+
+ l, err := cfgs[pkgName].Build(zp.AddCallerSkip(1))
+ if err != nil {
+ return nil, err
+ }
+
+ loggers[pkgName] = &logger{
+ log: l.Sugar(),
+ parent: l,
+ packageName: pkgName,
+ }
+ return loggers[pkgName], nil
+}
+
+//UpdateAllLoggers create new loggers for all registered pacakges with the defaultFields.
+func UpdateAllLoggers(defaultFields Fields) error {
+ for pkgName, cfg := range cfgs {
+ for k, v := range defaultFields {
+ if cfg.InitialFields == nil {
+ cfg.InitialFields = make(map[string]interface{})
+ }
+ cfg.InitialFields[k] = v
+ }
+ l, err := cfg.Build(zp.AddCallerSkip(1))
+ if err != nil {
+ return err
+ }
+
+ // Update the existing zap logger instance
+ loggers[pkgName].log = l.Sugar()
+ loggers[pkgName].parent = l
+ }
+ return nil
+}
+
+// Return a list of all packages that have individually-configured loggers
+func GetPackageNames() []string {
+ i := 0
+ keys := make([]string, len(loggers))
+ for k := range loggers {
+ keys[i] = k
+ i++
+ }
+ return keys
+}
+
+// UpdateLogger updates the logger associated with a caller's package with supplied defaultFields
+func UpdateLogger(defaultFields Fields) error {
+ pkgName, _, _, _ := getCallerInfo()
+ if _, exist := loggers[pkgName]; !exist {
+ return fmt.Errorf("package-%s-not-registered", pkgName)
+ }
+
+ // Build a new logger
+ if _, exist := cfgs[pkgName]; !exist {
+ return fmt.Errorf("config-%s-not-registered", pkgName)
+ }
+
+ cfg := cfgs[pkgName]
+ for k, v := range defaultFields {
+ if cfg.InitialFields == nil {
+ cfg.InitialFields = make(map[string]interface{})
+ }
+ cfg.InitialFields[k] = v
+ }
+ l, err := cfg.Build(zp.AddCallerSkip(1))
+ if err != nil {
+ return err
+ }
+
+ // Update the existing zap logger instance
+ loggers[pkgName].log = l.Sugar()
+ loggers[pkgName].parent = l
+
+ return nil
+}
+
+func setLevel(cfg zp.Config, level LogLevel) {
+ switch level {
+ case DebugLevel:
+ cfg.Level.SetLevel(zc.DebugLevel)
+ case InfoLevel:
+ cfg.Level.SetLevel(zc.InfoLevel)
+ case WarnLevel:
+ cfg.Level.SetLevel(zc.WarnLevel)
+ case ErrorLevel:
+ cfg.Level.SetLevel(zc.ErrorLevel)
+ case FatalLevel:
+ cfg.Level.SetLevel(zc.FatalLevel)
+ default:
+ cfg.Level.SetLevel(zc.ErrorLevel)
+ }
+}
+
+//SetPackageLogLevel dynamically sets the log level of a given package to level. This is typically invoked at an
+// application level during debugging
+func SetPackageLogLevel(packageName string, level LogLevel) {
+ // Get proper config
+ if cfg, ok := cfgs[packageName]; ok {
+ setLevel(cfg, level)
+ }
+}
+
+//SetAllLogLevel sets the log level of all registered packages to level
+func SetAllLogLevel(level LogLevel) {
+ // Get proper config
+ for _, cfg := range cfgs {
+ setLevel(cfg, level)
+ }
+}
+
+//GetPackageLogLevel returns the current log level of a package.
+func GetPackageLogLevel(packageName ...string) (LogLevel, error) {
+ var name string
+ if len(packageName) == 1 {
+ name = packageName[0]
+ } else {
+ name, _, _, _ = getCallerInfo()
+ }
+ if cfg, ok := cfgs[name]; ok {
+ return levelToLogLevel(cfg.Level.Level()), nil
+ }
+ return 0, fmt.Errorf("unknown-package-%s", name)
+}
+
+//GetDefaultLogLevel gets the log level used for packages that don't have specific loggers
+func GetDefaultLogLevel() LogLevel {
+ return levelToLogLevel(cfg.Level.Level())
+}
+
+//SetLogLevel sets the log level for the logger corresponding to the caller's package
+func SetLogLevel(level LogLevel) error {
+ pkgName, _, _, _ := getCallerInfo()
+ if _, exist := cfgs[pkgName]; !exist {
+ return fmt.Errorf("unregistered-package-%s", pkgName)
+ }
+ cfg := cfgs[pkgName]
+ setLevel(cfg, level)
+ return nil
+}
+
+//SetDefaultLogLevel sets the log level used for packages that don't have specific loggers
+func SetDefaultLogLevel(level LogLevel) {
+ setLevel(cfg, level)
+}
+
+// CleanUp flushed any buffered log entries. Applications should take care to call
+// CleanUp before exiting.
+func CleanUp() error {
+ for _, logger := range loggers {
+ if logger != nil {
+ if logger.parent != nil {
+ if err := logger.parent.Sync(); err != nil {
+ return err
+ }
+ }
+ }
+ }
+ if defaultLogger != nil {
+ if defaultLogger.parent != nil {
+ if err := defaultLogger.parent.Sync(); err != nil {
+ return err
+ }
+ }
+ }
+ return nil
+}
+
+func getCallerInfo() (string, string, string, int) {
+ // Since the caller of a log function is one stack frame before (in terms of stack higher level) the log.go
+ // filename, then first look for the last log.go filename and then grab the caller info one level higher.
+ maxLevel := 3
+ skiplevel := 3 // Level with the most empirical success to see the last log.go stack frame.
+ pc := make([]uintptr, maxLevel)
+ n := runtime.Callers(skiplevel, pc)
+ packageName := ""
+ funcName := ""
+ fileName := ""
+ var line int
+ if n == 0 {
+ return packageName, fileName, funcName, line
+ }
+ frames := runtime.CallersFrames(pc[:n])
+ var frame runtime.Frame
+ var foundFrame runtime.Frame
+ more := true
+ for more {
+ frame, more = frames.Next()
+ _, fileName = path.Split(frame.File)
+ if fileName != "log.go" {
+ foundFrame = frame // First frame after log.go in the frame stack
+ break
+ }
+ }
+ parts := strings.Split(foundFrame.Function, ".")
+ pl := len(parts)
+ if pl >= 2 {
+ funcName = parts[pl-1]
+ if parts[pl-2][0] == '(' {
+ packageName = strings.Join(parts[0:pl-2], ".")
+ } else {
+ packageName = strings.Join(parts[0:pl-1], ".")
+ }
+ }
+
+ if strings.HasSuffix(packageName, ".init") {
+ packageName = strings.TrimSuffix(packageName, ".init")
+ }
+
+ if strings.HasSuffix(fileName, ".go") {
+ fileName = strings.TrimSuffix(fileName, ".go")
+ }
+
+ return packageName, fileName, funcName, foundFrame.Line
+}
+
+func getPackageLevelSugaredLogger() *zp.SugaredLogger {
+ pkgName, fileName, funcName, line := getCallerInfo()
+ if _, exist := loggers[pkgName]; exist {
+ return loggers[pkgName].log.With("caller", fmt.Sprintf("%s.%s:%d", fileName, funcName, line))
+ }
+ return defaultLogger.log.With("caller", fmt.Sprintf("%s.%s:%d", fileName, funcName, line))
+}
+
+func getPackageLevelLogger() Logger {
+ pkgName, _, _, _ := getCallerInfo()
+ if _, exist := loggers[pkgName]; exist {
+ return loggers[pkgName]
+ }
+ return defaultLogger
+}
+
+func serializeMap(fields Fields) []interface{} {
+ data := make([]interface{}, len(fields)*2)
+ i := 0
+ for k, v := range fields {
+ data[i] = k
+ data[i+1] = v
+ i = i + 2
+ }
+ return data
+}
+
+// With returns a logger initialized with the key-value pairs
+func (l logger) With(keysAndValues Fields) Logger {
+ return logger{log: l.log.With(serializeMap(keysAndValues)...), parent: l.parent}
+}
+
+// Debug logs a message at level Debug on the standard logger.
+func (l logger) Debug(args ...interface{}) {
+ l.log.Debug(args...)
+}
+
+// Debugln logs a message at level Debug on the standard logger with a line feed. Default in any case.
+func (l logger) Debugln(args ...interface{}) {
+ l.log.Debug(args...)
+}
+
+// Debugw logs a message at level Debug on the standard logger.
+func (l logger) Debugf(format string, args ...interface{}) {
+ l.log.Debugf(format, args...)
+}
+
+// Debugw logs a message with some additional context. The variadic key-value
+// pairs are treated as they are in With.
+func (l logger) Debugw(msg string, keysAndValues Fields) {
+ l.log.Debugw(msg, serializeMap(keysAndValues)...)
+}
+
+// Info logs a message at level Info on the standard logger.
+func (l logger) Info(args ...interface{}) {
+ l.log.Info(args...)
+}
+
+// Infoln logs a message at level Info on the standard logger with a line feed. Default in any case.
+func (l logger) Infoln(args ...interface{}) {
+ l.log.Info(args...)
+ //msg := fmt.Sprintln(args...)
+ //l.sourced().Info(msg[:len(msg)-1])
+}
+
+// Infof logs a message at level Info on the standard logger.
+func (l logger) Infof(format string, args ...interface{}) {
+ l.log.Infof(format, args...)
+}
+
+// Infow logs a message with some additional context. The variadic key-value
+// pairs are treated as they are in With.
+func (l logger) Infow(msg string, keysAndValues Fields) {
+ l.log.Infow(msg, serializeMap(keysAndValues)...)
+}
+
+// Warn logs a message at level Warn on the standard logger.
+func (l logger) Warn(args ...interface{}) {
+ l.log.Warn(args...)
+}
+
+// Warnln logs a message at level Warn on the standard logger with a line feed. Default in any case.
+func (l logger) Warnln(args ...interface{}) {
+ l.log.Warn(args...)
+}
+
+// Warnf logs a message at level Warn on the standard logger.
+func (l logger) Warnf(format string, args ...interface{}) {
+ l.log.Warnf(format, args...)
+}
+
+// Warnw logs a message with some additional context. The variadic key-value
+// pairs are treated as they are in With.
+func (l logger) Warnw(msg string, keysAndValues Fields) {
+ l.log.Warnw(msg, serializeMap(keysAndValues)...)
+}
+
+// Error logs a message at level Error on the standard logger.
+func (l logger) Error(args ...interface{}) {
+ l.log.Error(args...)
+}
+
+// Errorln logs a message at level Error on the standard logger with a line feed. Default in any case.
+func (l logger) Errorln(args ...interface{}) {
+ l.log.Error(args...)
+}
+
+// Errorf logs a message at level Error on the standard logger.
+func (l logger) Errorf(format string, args ...interface{}) {
+ l.log.Errorf(format, args...)
+}
+
+// Errorw logs a message with some additional context. The variadic key-value
+// pairs are treated as they are in With.
+func (l logger) Errorw(msg string, keysAndValues Fields) {
+ l.log.Errorw(msg, serializeMap(keysAndValues)...)
+}
+
+// Fatal logs a message at level Fatal on the standard logger.
+func (l logger) Fatal(args ...interface{}) {
+ l.log.Fatal(args...)
+}
+
+// Fatalln logs a message at level Fatal on the standard logger with a line feed. Default in any case.
+func (l logger) Fatalln(args ...interface{}) {
+ l.log.Fatal(args...)
+}
+
+// Fatalf logs a message at level Fatal on the standard logger.
+func (l logger) Fatalf(format string, args ...interface{}) {
+ l.log.Fatalf(format, args...)
+}
+
+// Fatalw logs a message with some additional context. The variadic key-value
+// pairs are treated as they are in With.
+func (l logger) Fatalw(msg string, keysAndValues Fields) {
+ l.log.Fatalw(msg, serializeMap(keysAndValues)...)
+}
+
+// Warning logs a message at level Warn on the standard logger.
+func (l logger) Warning(args ...interface{}) {
+ l.log.Warn(args...)
+}
+
+// Warningln logs a message at level Warn on the standard logger with a line feed. Default in any case.
+func (l logger) Warningln(args ...interface{}) {
+ l.log.Warn(args...)
+}
+
+// Warningf logs a message at level Warn on the standard logger.
+func (l logger) Warningf(format string, args ...interface{}) {
+ l.log.Warnf(format, args...)
+}
+
+// V reports whether verbosity level l is at least the requested verbose level.
+func (l logger) V(level LogLevel) bool {
+ return l.parent.Core().Enabled(logLevelToLevel(level))
+}
+
+// GetLogLevel returns the current level of the logger
+func (l logger) GetLogLevel() LogLevel {
+ return levelToLogLevel(cfgs[l.packageName].Level.Level())
+}
+
+// With returns a logger initialized with the key-value pairs
+func With(keysAndValues Fields) Logger {
+ return logger{log: getPackageLevelSugaredLogger().With(serializeMap(keysAndValues)...), parent: defaultLogger.parent}
+}
+
+// Debug logs a message at level Debug on the standard logger.
+func Debug(args ...interface{}) {
+ getPackageLevelSugaredLogger().Debug(args...)
+}
+
+// Debugln logs a message at level Debug on the standard logger.
+func Debugln(args ...interface{}) {
+ getPackageLevelSugaredLogger().Debug(args...)
+}
+
+// Debugf logs a message at level Debug on the standard logger.
+func Debugf(format string, args ...interface{}) {
+ getPackageLevelSugaredLogger().Debugf(format, args...)
+}
+
+// Debugw logs a message with some additional context. The variadic key-value
+// pairs are treated as they are in With.
+func Debugw(msg string, keysAndValues Fields) {
+ getPackageLevelSugaredLogger().Debugw(msg, serializeMap(keysAndValues)...)
+}
+
+// Info logs a message at level Info on the standard logger.
+func Info(args ...interface{}) {
+ getPackageLevelSugaredLogger().Info(args...)
+}
+
+// Infoln logs a message at level Info on the standard logger.
+func Infoln(args ...interface{}) {
+ getPackageLevelSugaredLogger().Info(args...)
+}
+
+// Infof logs a message at level Info on the standard logger.
+func Infof(format string, args ...interface{}) {
+ getPackageLevelSugaredLogger().Infof(format, args...)
+}
+
+//Infow logs a message with some additional context. The variadic key-value
+//pairs are treated as they are in With.
+func Infow(msg string, keysAndValues Fields) {
+ getPackageLevelSugaredLogger().Infow(msg, serializeMap(keysAndValues)...)
+}
+
+// Warn logs a message at level Warn on the standard logger.
+func Warn(args ...interface{}) {
+ getPackageLevelSugaredLogger().Warn(args...)
+}
+
+// Warnln logs a message at level Warn on the standard logger.
+func Warnln(args ...interface{}) {
+ getPackageLevelSugaredLogger().Warn(args...)
+}
+
+// Warnf logs a message at level Warn on the standard logger.
+func Warnf(format string, args ...interface{}) {
+ getPackageLevelSugaredLogger().Warnf(format, args...)
+}
+
+// Warnw logs a message with some additional context. The variadic key-value
+// pairs are treated as they are in With.
+func Warnw(msg string, keysAndValues Fields) {
+ getPackageLevelSugaredLogger().Warnw(msg, serializeMap(keysAndValues)...)
+}
+
+// Error logs a message at level Error on the standard logger.
+func Error(args ...interface{}) {
+ getPackageLevelSugaredLogger().Error(args...)
+}
+
+// Errorln logs a message at level Error on the standard logger.
+func Errorln(args ...interface{}) {
+ getPackageLevelSugaredLogger().Error(args...)
+}
+
+// Errorf logs a message at level Error on the standard logger.
+func Errorf(format string, args ...interface{}) {
+ getPackageLevelSugaredLogger().Errorf(format, args...)
+}
+
+// Errorw logs a message with some additional context. The variadic key-value
+// pairs are treated as they are in With.
+func Errorw(msg string, keysAndValues Fields) {
+ getPackageLevelSugaredLogger().Errorw(msg, serializeMap(keysAndValues)...)
+}
+
+// Fatal logs a message at level Fatal on the standard logger.
+func Fatal(args ...interface{}) {
+ getPackageLevelSugaredLogger().Fatal(args...)
+}
+
+// Fatalln logs a message at level Fatal on the standard logger.
+func Fatalln(args ...interface{}) {
+ getPackageLevelSugaredLogger().Fatal(args...)
+}
+
+// Fatalf logs a message at level Fatal on the standard logger.
+func Fatalf(format string, args ...interface{}) {
+ getPackageLevelSugaredLogger().Fatalf(format, args...)
+}
+
+// Fatalw logs a message with some additional context. The variadic key-value
+// pairs are treated as they are in With.
+func Fatalw(msg string, keysAndValues Fields) {
+ getPackageLevelSugaredLogger().Fatalw(msg, serializeMap(keysAndValues)...)
+}
+
+// Warning logs a message at level Warn on the standard logger.
+func Warning(args ...interface{}) {
+ getPackageLevelSugaredLogger().Warn(args...)
+}
+
+// Warningln logs a message at level Warn on the standard logger.
+func Warningln(args ...interface{}) {
+ getPackageLevelSugaredLogger().Warn(args...)
+}
+
+// Warningf logs a message at level Warn on the standard logger.
+func Warningf(format string, args ...interface{}) {
+ getPackageLevelSugaredLogger().Warnf(format, args...)
+}
+
+// V reports whether verbosity level l is at least the requested verbose level.
+func V(level LogLevel) bool {
+ return getPackageLevelLogger().V(level)
+}
+
+//GetLogLevel returns the log level of the invoking package
+func GetLogLevel() LogLevel {
+ return getPackageLevelLogger().GetLogLevel()
+}
diff --git a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/ponresourcemanager/ponresourcemanager.go b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/ponresourcemanager/ponresourcemanager.go
new file mode 100644
index 0000000..79fefc5
--- /dev/null
+++ b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/ponresourcemanager/ponresourcemanager.go
@@ -0,0 +1,1337 @@
+/*
+ * Copyright 2019-present Open Networking Foundation
+
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+
+ * http://www.apache.org/licenses/LICENSE-2.0
+
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package ponresourcemanager
+
+import (
+ "context"
+ "encoding/base64"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "strconv"
+
+ bitmap "github.com/boljen/go-bitmap"
+ "github.com/opencord/voltha-lib-go/v3/pkg/db"
+ "github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore"
+ "github.com/opencord/voltha-lib-go/v3/pkg/log"
+ tp "github.com/opencord/voltha-lib-go/v3/pkg/techprofile"
+)
+
+const (
+ //Constants to identify resource pool
+ UNI_ID = "UNI_ID"
+ ONU_ID = "ONU_ID"
+ ALLOC_ID = "ALLOC_ID"
+ GEMPORT_ID = "GEMPORT_ID"
+ FLOW_ID = "FLOW_ID"
+
+ //Constants for passing command line arugments
+ OLT_MODEL_ARG = "--olt_model"
+ PATH_PREFIX = "service/voltha/resource_manager/{%s}"
+
+ /*The path under which configuration data is stored is defined as technology/device agnostic.
+ That means the path does not include any specific technology/device variable. Using technology/device
+ agnostic path also makes northbound applications, that need to write to this path,
+ technology/device agnostic.
+
+ Default kv client of PonResourceManager reads from/writes to PATH_PREFIX defined above.
+ That is why, an additional kv client (named KVStoreForConfig) is defined to read from the config path.
+ */
+ PATH_PREFIX_FOR_CONFIG = "service/voltha/resource_manager/config"
+ /*The resource ranges for a given device model should be placed
+ at 'resource_manager/<technology>/resource_ranges/<olt_model_type>'
+ path on the KV store.
+ If Resource Range parameters are to be read from the external KV store,
+ they are expected to be stored in the following format.
+ Note: All parameters are MANDATORY for now.
+ constants used as keys to reference the resource range parameters from
+ and external KV store.
+ */
+ UNI_ID_START_IDX = "uni_id_start"
+ UNI_ID_END_IDX = "uni_id_end"
+ ONU_ID_START_IDX = "onu_id_start"
+ ONU_ID_END_IDX = "onu_id_end"
+ ONU_ID_SHARED_IDX = "onu_id_shared"
+ ALLOC_ID_START_IDX = "alloc_id_start"
+ ALLOC_ID_END_IDX = "alloc_id_end"
+ ALLOC_ID_SHARED_IDX = "alloc_id_shared"
+ GEMPORT_ID_START_IDX = "gemport_id_start"
+ GEMPORT_ID_END_IDX = "gemport_id_end"
+ GEMPORT_ID_SHARED_IDX = "gemport_id_shared"
+ FLOW_ID_START_IDX = "flow_id_start"
+ FLOW_ID_END_IDX = "flow_id_end"
+ FLOW_ID_SHARED_IDX = "flow_id_shared"
+ NUM_OF_PON_PORT = "pon_ports"
+
+ /*
+ The KV store backend is initialized with a path prefix and we need to
+ provide only the suffix.
+ */
+ PON_RESOURCE_RANGE_CONFIG_PATH = "resource_ranges/%s"
+
+ //resource path suffix
+ //Path on the KV store for storing alloc id ranges and resource pool for a given interface
+ //Format: <device_id>/alloc_id_pool/<pon_intf_id>
+ ALLOC_ID_POOL_PATH = "{%s}/alloc_id_pool/{%d}"
+ //Path on the KV store for storing gemport id ranges and resource pool for a given interface
+ //Format: <device_id>/gemport_id_pool/<pon_intf_id>
+ GEMPORT_ID_POOL_PATH = "{%s}/gemport_id_pool/{%d}"
+ //Path on the KV store for storing onu id ranges and resource pool for a given interface
+ //Format: <device_id>/onu_id_pool/<pon_intf_id>
+ ONU_ID_POOL_PATH = "{%s}/onu_id_pool/{%d}"
+ //Path on the KV store for storing flow id ranges and resource pool for a given interface
+ //Format: <device_id>/flow_id_pool/<pon_intf_id>
+ FLOW_ID_POOL_PATH = "{%s}/flow_id_pool/{%d}"
+
+ //Path on the KV store for storing list of alloc IDs for a given ONU
+ //Format: <device_id>/<(pon_intf_id, onu_id)>/alloc_ids
+ ALLOC_ID_RESOURCE_MAP_PATH = "{%s}/{%s}/alloc_ids"
+
+ //Path on the KV store for storing list of gemport IDs for a given ONU
+ //Format: <device_id>/<(pon_intf_id, onu_id)>/gemport_ids
+ GEMPORT_ID_RESOURCE_MAP_PATH = "{%s}/{%s}/gemport_ids"
+
+ //Path on the KV store for storing list of Flow IDs for a given ONU
+ //Format: <device_id>/<(pon_intf_id, onu_id)>/flow_ids
+ FLOW_ID_RESOURCE_MAP_PATH = "{%s}/{%s}/flow_ids"
+
+ //Flow Id info: Use to store more metadata associated with the flow_id
+ //Format: <device_id>/<(pon_intf_id, onu_id)>/flow_id_info/<flow_id>
+ FLOW_ID_INFO_PATH = "{%s}/{%s}/flow_id_info/{%d}"
+
+ //path on the kvstore to store onugem info map
+ //format: <device-id>/onu_gem_info/<intfid>
+ ONU_GEM_INFO_PATH = "{%s}/onu_gem_info/{%d}" // onu_gem/<(intfid)>
+
+ //Constants for internal usage.
+ PON_INTF_ID = "pon_intf_id"
+ START_IDX = "start_idx"
+ END_IDX = "end_idx"
+ POOL = "pool"
+ NUM_OF_PON_INTF = 16
+
+ KVSTORE_RETRY_TIMEOUT = 5
+ //Path on the KV store for storing reserved gem ports
+ //Format: reserved_gemport_ids
+ RESERVED_GEMPORT_IDS_PATH = "reserved_gemport_ids"
+)
+
+//type ResourceTypeIndex string
+//type ResourceType string
+
+type PONResourceManager struct {
+ //Implements APIs to initialize/allocate/release alloc/gemport/onu IDs.
+ Technology string
+ DeviceType string
+ DeviceID string
+ Backend string // ETCD, or consul
+ Host string // host ip of the KV store
+ Port int // port number for the KV store
+ OLTModel string
+ KVStore *db.Backend
+ KVStoreForConfig *db.Backend
+ TechProfileMgr tp.TechProfileIf // create object of *tp.TechProfileMgr
+
+ // Below attribute, pon_resource_ranges, should be initialized
+ // by reading from KV store.
+ PonResourceRanges map[string]interface{}
+ SharedResourceMgrs map[string]*PONResourceManager
+ SharedIdxByType map[string]string
+ IntfIDs []uint32 // list of pon interface IDs
+ Globalorlocal string
+}
+
+func newKVClient(storeType string, address string, timeout int) (kvstore.Client, error) {
+ log.Infow("kv-store-type", log.Fields{"store": storeType})
+ switch storeType {
+ case "consul":
+ return kvstore.NewConsulClient(address, timeout)
+ case "etcd":
+ return kvstore.NewEtcdClient(address, timeout)
+ }
+ return nil, errors.New("unsupported-kv-store")
+}
+
+func SetKVClient(Technology string, Backend string, Host string, Port int, configClient bool) *db.Backend {
+ addr := Host + ":" + strconv.Itoa(Port)
+ // TODO : Make sure direct call to NewBackend is working fine with backend , currently there is some
+ // issue between kv store and backend , core is not calling NewBackend directly
+ kvClient, err := newKVClient(Backend, addr, KVSTORE_RETRY_TIMEOUT)
+ if err != nil {
+ log.Fatalw("Failed to init KV client\n", log.Fields{"err": err})
+ return nil
+ }
+
+ var pathPrefix string
+ if configClient {
+ pathPrefix = PATH_PREFIX_FOR_CONFIG
+ } else {
+ pathPrefix = fmt.Sprintf(PATH_PREFIX, Technology)
+ }
+
+ kvbackend := &db.Backend{
+ Client: kvClient,
+ StoreType: Backend,
+ Host: Host,
+ Port: Port,
+ Timeout: KVSTORE_RETRY_TIMEOUT,
+ PathPrefix: pathPrefix}
+
+ return kvbackend
+}
+
+// NewPONResourceManager creates a new PON resource manager.
+func NewPONResourceManager(Technology string, DeviceType string, DeviceID string, Backend string, Host string, Port int) (*PONResourceManager, error) {
+ var PONMgr PONResourceManager
+ PONMgr.Technology = Technology
+ PONMgr.DeviceType = DeviceType
+ PONMgr.DeviceID = DeviceID
+ PONMgr.Backend = Backend
+ PONMgr.Host = Host
+ PONMgr.Port = Port
+ PONMgr.KVStore = SetKVClient(Technology, Backend, Host, Port, false)
+ if PONMgr.KVStore == nil {
+ log.Error("KV Client initilization failed")
+ return nil, errors.New("Failed to init KV client")
+ }
+ // init kv client to read from the config path
+ PONMgr.KVStoreForConfig = SetKVClient(Technology, Backend, Host, Port, true)
+ if PONMgr.KVStoreForConfig == nil {
+ log.Error("KV Config Client initilization failed")
+ return nil, errors.New("Failed to init KV Config client")
+ }
+ // Initialize techprofile for this technology
+ if PONMgr.TechProfileMgr, _ = tp.NewTechProfile(&PONMgr, Backend, Host, Port); PONMgr.TechProfileMgr == nil {
+ log.Error("Techprofile initialization failed")
+ return nil, errors.New("Failed to init tech profile")
+ }
+ PONMgr.PonResourceRanges = make(map[string]interface{})
+ PONMgr.SharedResourceMgrs = make(map[string]*PONResourceManager)
+ PONMgr.SharedIdxByType = make(map[string]string)
+ PONMgr.SharedIdxByType[ONU_ID] = ONU_ID_SHARED_IDX
+ PONMgr.SharedIdxByType[ALLOC_ID] = ALLOC_ID_SHARED_IDX
+ PONMgr.SharedIdxByType[GEMPORT_ID] = GEMPORT_ID_SHARED_IDX
+ PONMgr.SharedIdxByType[FLOW_ID] = FLOW_ID_SHARED_IDX
+ PONMgr.IntfIDs = make([]uint32, NUM_OF_PON_INTF)
+ PONMgr.OLTModel = DeviceType
+ return &PONMgr, nil
+}
+
+/*
+ Initialize PON resource ranges with config fetched from kv store.
+ return boolean: True if PON resource ranges initialized else false
+ Try to initialize the PON Resource Ranges from KV store based on the
+ OLT model key, if available
+*/
+
+func (PONRMgr *PONResourceManager) InitResourceRangesFromKVStore(ctx context.Context) bool {
+ //Initialize PON resource ranges with config fetched from kv store.
+ //:return boolean: True if PON resource ranges initialized else false
+ // Try to initialize the PON Resource Ranges from KV store based on the
+ // OLT model key, if available
+ if PONRMgr.OLTModel == "" {
+ log.Error("Failed to get OLT model")
+ return false
+ }
+ Path := fmt.Sprintf(PON_RESOURCE_RANGE_CONFIG_PATH, PONRMgr.OLTModel)
+ //get resource from kv store
+ Result, err := PONRMgr.KVStore.Get(ctx, Path)
+ if err != nil {
+ log.Debugf("Error in fetching resource %s from KV strore", Path)
+ return false
+ }
+ if Result == nil {
+ log.Debug("There may be no resources in the KV store in case of fresh bootup, return true")
+ return false
+ }
+ //update internal ranges from kv ranges. If there are missing
+ // values in the KV profile, continue to use the defaults
+ Value, err := ToByte(Result.Value)
+ if err != nil {
+ log.Error("Failed to convert kvpair to byte string")
+ return false
+ }
+ if err := json.Unmarshal(Value, &PONRMgr.PonResourceRanges); err != nil {
+ log.Error("Failed to Unmarshal json byte")
+ return false
+ }
+ log.Debug("Init resource ranges from kvstore success")
+ return true
+}
+
+func (PONRMgr *PONResourceManager) UpdateRanges(StartIDx string, StartID uint32, EndIDx string, EndID uint32,
+ SharedIDx string, SharedPoolID uint32, RMgr *PONResourceManager) {
+ /*
+ Update the ranges for all reosurce type in the intermnal maps
+ param: resource type start index
+ param: start ID
+ param: resource type end index
+ param: end ID
+ param: resource type shared index
+ param: shared pool id
+ param: global resource manager
+ */
+ log.Debugf("update ranges for %s, %d", StartIDx, StartID)
+
+ if StartID != 0 {
+ if (PONRMgr.PonResourceRanges[StartIDx] == nil) || (PONRMgr.PonResourceRanges[StartIDx].(uint32) < StartID) {
+ PONRMgr.PonResourceRanges[StartIDx] = StartID
+ }
+ }
+ if EndID != 0 {
+ if (PONRMgr.PonResourceRanges[EndIDx] == nil) || (PONRMgr.PonResourceRanges[EndIDx].(uint32) > EndID) {
+ PONRMgr.PonResourceRanges[EndIDx] = EndID
+ }
+ }
+ //if SharedPoolID != 0 {
+ PONRMgr.PonResourceRanges[SharedIDx] = SharedPoolID
+ //}
+ if RMgr != nil {
+ PONRMgr.SharedResourceMgrs[SharedIDx] = RMgr
+ }
+}
+
+func (PONRMgr *PONResourceManager) InitDefaultPONResourceRanges(ONUIDStart uint32,
+ ONUIDEnd uint32,
+ ONUIDSharedPoolID uint32,
+ AllocIDStart uint32,
+ AllocIDEnd uint32,
+ AllocIDSharedPoolID uint32,
+ GEMPortIDStart uint32,
+ GEMPortIDEnd uint32,
+ GEMPortIDSharedPoolID uint32,
+ FlowIDStart uint32,
+ FlowIDEnd uint32,
+ FlowIDSharedPoolID uint32,
+ UNIIDStart uint32,
+ UNIIDEnd uint32,
+ NoOfPONPorts uint32,
+ IntfIDs []uint32) bool {
+
+ /*Initialize default PON resource ranges
+
+ :param onu_id_start_idx: onu id start index
+ :param onu_id_end_idx: onu id end index
+ :param onu_id_shared_pool_id: pool idx for id shared by all intfs or None for no sharing
+ :param alloc_id_start_idx: alloc id start index
+ :param alloc_id_end_idx: alloc id end index
+ :param alloc_id_shared_pool_id: pool idx for alloc id shared by all intfs or None for no sharing
+ :param gemport_id_start_idx: gemport id start index
+ :param gemport_id_end_idx: gemport id end index
+ :param gemport_id_shared_pool_id: pool idx for gemport id shared by all intfs or None for no sharing
+ :param flow_id_start_idx: flow id start index
+ :param flow_id_end_idx: flow id end index
+ :param flow_id_shared_pool_id: pool idx for flow id shared by all intfs or None for no sharing
+ :param num_of_pon_ports: number of PON ports
+ :param intf_ids: interfaces serviced by this manager
+ */
+ PONRMgr.UpdateRanges(ONU_ID_START_IDX, ONUIDStart, ONU_ID_END_IDX, ONUIDEnd, ONU_ID_SHARED_IDX, ONUIDSharedPoolID, nil)
+ PONRMgr.UpdateRanges(ALLOC_ID_START_IDX, AllocIDStart, ALLOC_ID_END_IDX, AllocIDEnd, ALLOC_ID_SHARED_IDX, AllocIDSharedPoolID, nil)
+ PONRMgr.UpdateRanges(GEMPORT_ID_START_IDX, GEMPortIDStart, GEMPORT_ID_END_IDX, GEMPortIDEnd, GEMPORT_ID_SHARED_IDX, GEMPortIDSharedPoolID, nil)
+ PONRMgr.UpdateRanges(FLOW_ID_START_IDX, FlowIDStart, FLOW_ID_END_IDX, FlowIDEnd, FLOW_ID_SHARED_IDX, FlowIDSharedPoolID, nil)
+ PONRMgr.UpdateRanges(UNI_ID_START_IDX, UNIIDStart, UNI_ID_END_IDX, UNIIDEnd, "", 0, nil)
+ log.Debug("Initialize default range values")
+ var i uint32
+ if IntfIDs == nil {
+ for i = 0; i < NoOfPONPorts; i++ {
+ PONRMgr.IntfIDs = append(PONRMgr.IntfIDs, i)
+ }
+ } else {
+ PONRMgr.IntfIDs = IntfIDs
+ }
+ return true
+}
+
+func (PONRMgr *PONResourceManager) InitDeviceResourcePool(ctx context.Context) error {
+
+ //Initialize resource pool for all PON ports.
+
+ log.Debug("Init resource ranges")
+
+ var err error
+ for _, Intf := range PONRMgr.IntfIDs {
+ SharedPoolID := PONRMgr.PonResourceRanges[ONU_ID_SHARED_IDX].(uint32)
+ if SharedPoolID != 0 {
+ Intf = SharedPoolID
+ }
+ if err = PONRMgr.InitResourceIDPool(ctx, Intf, ONU_ID,
+ PONRMgr.PonResourceRanges[ONU_ID_START_IDX].(uint32),
+ PONRMgr.PonResourceRanges[ONU_ID_END_IDX].(uint32)); err != nil {
+ log.Error("Failed to init ONU ID resource pool")
+ return err
+ }
+ if SharedPoolID != 0 {
+ break
+ }
+ }
+
+ for _, Intf := range PONRMgr.IntfIDs {
+ SharedPoolID := PONRMgr.PonResourceRanges[ALLOC_ID_SHARED_IDX].(uint32)
+ if SharedPoolID != 0 {
+ Intf = SharedPoolID
+ }
+ if err = PONRMgr.InitResourceIDPool(ctx, Intf, ALLOC_ID,
+ PONRMgr.PonResourceRanges[ALLOC_ID_START_IDX].(uint32),
+ PONRMgr.PonResourceRanges[ALLOC_ID_END_IDX].(uint32)); err != nil {
+ log.Error("Failed to init ALLOC ID resource pool ")
+ return err
+ }
+ if SharedPoolID != 0 {
+ break
+ }
+ }
+ for _, Intf := range PONRMgr.IntfIDs {
+ SharedPoolID := PONRMgr.PonResourceRanges[GEMPORT_ID_SHARED_IDX].(uint32)
+ if SharedPoolID != 0 {
+ Intf = SharedPoolID
+ }
+ if err = PONRMgr.InitResourceIDPool(ctx, Intf, GEMPORT_ID,
+ PONRMgr.PonResourceRanges[GEMPORT_ID_START_IDX].(uint32),
+ PONRMgr.PonResourceRanges[GEMPORT_ID_END_IDX].(uint32)); err != nil {
+ log.Error("Failed to init GEMPORT ID resource pool")
+ return err
+ }
+ if SharedPoolID != 0 {
+ break
+ }
+ }
+
+ for _, Intf := range PONRMgr.IntfIDs {
+ SharedPoolID := PONRMgr.PonResourceRanges[FLOW_ID_SHARED_IDX].(uint32)
+ if SharedPoolID != 0 {
+ Intf = SharedPoolID
+ }
+ if err = PONRMgr.InitResourceIDPool(ctx, Intf, FLOW_ID,
+ PONRMgr.PonResourceRanges[FLOW_ID_START_IDX].(uint32),
+ PONRMgr.PonResourceRanges[FLOW_ID_END_IDX].(uint32)); err != nil {
+ log.Error("Failed to init FLOW ID resource pool")
+ return err
+ }
+ if SharedPoolID != 0 {
+ break
+ }
+ }
+ return err
+}
+
+func (PONRMgr *PONResourceManager) ClearDeviceResourcePool(ctx context.Context) error {
+
+ //Clear resource pool for all PON ports.
+
+ log.Debug("Clear resource ranges")
+
+ for _, Intf := range PONRMgr.IntfIDs {
+ SharedPoolID := PONRMgr.PonResourceRanges[ONU_ID_SHARED_IDX].(uint32)
+ if SharedPoolID != 0 {
+ Intf = SharedPoolID
+ }
+ if status := PONRMgr.ClearResourceIDPool(ctx, Intf, ONU_ID); !status {
+ log.Error("Failed to clear ONU ID resource pool")
+ return errors.New("Failed to clear ONU ID resource pool")
+ }
+ if SharedPoolID != 0 {
+ break
+ }
+ }
+
+ for _, Intf := range PONRMgr.IntfIDs {
+ SharedPoolID := PONRMgr.PonResourceRanges[ALLOC_ID_SHARED_IDX].(uint32)
+ if SharedPoolID != 0 {
+ Intf = SharedPoolID
+ }
+ if status := PONRMgr.ClearResourceIDPool(ctx, Intf, ALLOC_ID); !status {
+ log.Error("Failed to clear ALLOC ID resource pool ")
+ return errors.New("Failed to clear ALLOC ID resource pool")
+ }
+ if SharedPoolID != 0 {
+ break
+ }
+ }
+ for _, Intf := range PONRMgr.IntfIDs {
+ SharedPoolID := PONRMgr.PonResourceRanges[GEMPORT_ID_SHARED_IDX].(uint32)
+ if SharedPoolID != 0 {
+ Intf = SharedPoolID
+ }
+ if status := PONRMgr.ClearResourceIDPool(ctx, Intf, GEMPORT_ID); !status {
+ log.Error("Failed to clear GEMPORT ID resource pool")
+ return errors.New("Failed to clear GEMPORT ID resource pool")
+ }
+ if SharedPoolID != 0 {
+ break
+ }
+ }
+
+ for _, Intf := range PONRMgr.IntfIDs {
+ SharedPoolID := PONRMgr.PonResourceRanges[FLOW_ID_SHARED_IDX].(uint32)
+ if SharedPoolID != 0 {
+ Intf = SharedPoolID
+ }
+ if status := PONRMgr.ClearResourceIDPool(ctx, Intf, FLOW_ID); !status {
+ log.Error("Failed to clear FLOW ID resource pool")
+ return errors.New("Failed to clear FLOW ID resource pool")
+ }
+ if SharedPoolID != 0 {
+ break
+ }
+ }
+ return nil
+}
+
+func (PONRMgr *PONResourceManager) InitResourceIDPool(ctx context.Context, Intf uint32, ResourceType string, StartID uint32, EndID uint32) error {
+
+ /*Initialize Resource ID pool for a given Resource Type on a given PON Port
+
+ :param pon_intf_id: OLT PON interface id
+ :param resource_type: String to identify type of resource
+ :param start_idx: start index for onu id pool
+ :param end_idx: end index for onu id pool
+ :return boolean: True if resource id pool initialized else false
+ */
+
+ // delegate to the master instance if sharing enabled across instances
+ SharedResourceMgr := PONRMgr.SharedResourceMgrs[PONRMgr.SharedIdxByType[ResourceType]]
+ if SharedResourceMgr != nil && PONRMgr != SharedResourceMgr {
+ return SharedResourceMgr.InitResourceIDPool(ctx, Intf, ResourceType, StartID, EndID)
+ }
+
+ Path := PONRMgr.GetPath(Intf, ResourceType)
+ if Path == "" {
+ log.Errorf("Failed to get path for resource type %s", ResourceType)
+ return fmt.Errorf("Failed to get path for resource type %s", ResourceType)
+ }
+
+ //In case of adapter reboot and reconciliation resource in kv store
+ //checked for its presence if not kv store update happens
+ Res, err := PONRMgr.GetResource(ctx, Path)
+ if (err == nil) && (Res != nil) {
+ log.Debugf("Resource %s already present in store ", Path)
+ return nil
+ } else {
+ var excluded []uint32
+ if ResourceType == GEMPORT_ID {
+ //get gem port ids defined in the KV store, if any, and exclude them from the gem port id pool
+ if reservedGemPortIds, defined := PONRMgr.getReservedGemPortIdsFromKVStore(ctx); defined {
+ excluded = reservedGemPortIds
+ log.Debugw("Excluding some ports from GEM port id pool", log.Fields{"excluded gem ports": excluded})
+ }
+ }
+ FormatResult, err := PONRMgr.FormatResource(Intf, StartID, EndID, excluded)
+ if err != nil {
+ log.Errorf("Failed to format resource")
+ return err
+ }
+ // Add resource as json in kv store.
+ err = PONRMgr.KVStore.Put(ctx, Path, FormatResult)
+ if err == nil {
+ log.Debug("Successfuly posted to kv store")
+ return err
+ }
+ }
+
+ log.Debug("Error initializing pool")
+
+ return err
+}
+
+func (PONRMgr *PONResourceManager) getReservedGemPortIdsFromKVStore(ctx context.Context) ([]uint32, bool) {
+ var reservedGemPortIds []uint32
+ // read reserved gem ports from the config path
+ KvPair, err := PONRMgr.KVStoreForConfig.Get(ctx, RESERVED_GEMPORT_IDS_PATH)
+ if err != nil {
+ log.Errorw("Unable to get reserved GEM port ids from the kv store", log.Fields{"err": err})
+ return reservedGemPortIds, false
+ }
+ if KvPair == nil || KvPair.Value == nil {
+ //no reserved gem port defined in the store
+ return reservedGemPortIds, false
+ }
+ Val, err := kvstore.ToByte(KvPair.Value)
+ if err != nil {
+ log.Errorw("Failed to convert reserved gem port ids into byte array", log.Fields{"err": err})
+ return reservedGemPortIds, false
+ }
+ if err = json.Unmarshal(Val, &reservedGemPortIds); err != nil {
+ log.Errorw("Failed to unmarshal reservedGemPortIds", log.Fields{"err": err})
+ return reservedGemPortIds, false
+ }
+ return reservedGemPortIds, true
+}
+
+func (PONRMgr *PONResourceManager) FormatResource(IntfID uint32, StartIDx uint32, EndIDx uint32,
+ Excluded []uint32) ([]byte, error) {
+ /*
+ Format resource as json.
+ :param pon_intf_id: OLT PON interface id
+ :param start_idx: start index for id pool
+ :param end_idx: end index for id pool
+ :Id values to be Excluded from the pool
+ :return dictionary: resource formatted as map
+ */
+ // Format resource as json to be stored in backend store
+ Resource := make(map[string]interface{})
+ Resource[PON_INTF_ID] = IntfID
+ Resource[START_IDX] = StartIDx
+ Resource[END_IDX] = EndIDx
+ /*
+ Resource pool stored in backend store as binary string.
+ Tracking the resource allocation will be done by setting the bits \
+ in the byte array. The index set will be the resource number allocated.
+ */
+ var TSData *bitmap.Threadsafe
+ if TSData = bitmap.NewTS(int(EndIDx)); TSData == nil {
+ log.Error("Failed to create a bitmap")
+ return nil, errors.New("Failed to create bitmap")
+ }
+ for _, excludedID := range Excluded {
+ if excludedID < StartIDx || excludedID > EndIDx {
+ log.Warnf("Cannot reserve %d. It must be in the range of [%d, %d]", excludedID,
+ StartIDx, EndIDx)
+ continue
+ }
+ PONRMgr.reserveID(TSData, StartIDx, excludedID)
+ }
+ Resource[POOL] = TSData.Data(false) //we pass false so as the TSData lib api does not do a copy of the data and return
+
+ Value, err := json.Marshal(Resource)
+ if err != nil {
+ log.Errorf("Failed to marshall resource")
+ return nil, err
+ }
+ return Value, err
+}
+func (PONRMgr *PONResourceManager) GetResource(ctx context.Context, Path string) (map[string]interface{}, error) {
+ /*
+ Get resource from kv store.
+
+ :param path: path to get resource
+ :return: resource if resource present in kv store else None
+ */
+ //get resource from kv store
+
+ var Value []byte
+ Result := make(map[string]interface{})
+ var Str string
+
+ Resource, err := PONRMgr.KVStore.Get(ctx, Path)
+ if (err != nil) || (Resource == nil) {
+ log.Debugf("Resource unavailable at %s", Path)
+ return nil, err
+ }
+
+ Value, err = ToByte(Resource.Value)
+ if err != nil {
+ return nil, err
+ }
+
+ // decode resource fetched from backend store to dictionary
+ err = json.Unmarshal(Value, &Result)
+ if err != nil {
+ log.Error("Failed to decode resource")
+ return Result, err
+ }
+ /*
+ resource pool in backend store stored as binary string whereas to
+ access the pool to generate/release IDs it need to be converted
+ as BitArray
+ */
+ Str, err = ToString(Result[POOL])
+ if err != nil {
+ log.Error("Failed to conver to kv pair to string")
+ return Result, err
+ }
+ Decode64, _ := base64.StdEncoding.DecodeString(Str)
+ Result[POOL], err = ToByte(Decode64)
+ if err != nil {
+ log.Error("Failed to convert resource pool to byte")
+ return Result, err
+ }
+
+ return Result, err
+}
+
+func (PONRMgr *PONResourceManager) GetPath(IntfID uint32, ResourceType string) string {
+ /*
+ Get path for given resource type.
+ :param pon_intf_id: OLT PON interface id
+ :param resource_type: String to identify type of resource
+ :return: path for given resource type
+ */
+
+ /*
+ Get the shared pool for the given resource type.
+ all the resource ranges and the shared resource maps are initialized during the init.
+ */
+ SharedPoolID := PONRMgr.PonResourceRanges[PONRMgr.SharedIdxByType[ResourceType]].(uint32)
+ if SharedPoolID != 0 {
+ IntfID = SharedPoolID
+ }
+ var Path string
+ if ResourceType == ONU_ID {
+ Path = fmt.Sprintf(ONU_ID_POOL_PATH, PONRMgr.DeviceID, IntfID)
+ } else if ResourceType == ALLOC_ID {
+ Path = fmt.Sprintf(ALLOC_ID_POOL_PATH, PONRMgr.DeviceID, IntfID)
+ } else if ResourceType == GEMPORT_ID {
+ Path = fmt.Sprintf(GEMPORT_ID_POOL_PATH, PONRMgr.DeviceID, IntfID)
+ } else if ResourceType == FLOW_ID {
+ Path = fmt.Sprintf(FLOW_ID_POOL_PATH, PONRMgr.DeviceID, IntfID)
+ } else {
+ log.Error("Invalid resource pool identifier")
+ }
+ return Path
+}
+
+func (PONRMgr *PONResourceManager) GetResourceID(ctx context.Context, IntfID uint32, ResourceType string, NumIDs uint32) ([]uint32, error) {
+ /*
+ Create alloc/gemport/onu/flow id for given OLT PON interface.
+ :param pon_intf_id: OLT PON interface id
+ :param resource_type: String to identify type of resource
+ :param num_of_id: required number of ids
+ :return list/uint32/None: list, uint32 or None if resource type is
+ alloc_id/gemport_id, onu_id or invalid type respectively
+ */
+ if NumIDs < 1 {
+ log.Error("Invalid number of resources requested")
+ return nil, fmt.Errorf("Invalid number of resources requested %d", NumIDs)
+ }
+ // delegate to the master instance if sharing enabled across instances
+
+ SharedResourceMgr := PONRMgr.SharedResourceMgrs[PONRMgr.SharedIdxByType[ResourceType]]
+ if SharedResourceMgr != nil && PONRMgr != SharedResourceMgr {
+ return SharedResourceMgr.GetResourceID(ctx, IntfID, ResourceType, NumIDs)
+ }
+ log.Debugf("Fetching resource from %s rsrc mgr for resource %s", PONRMgr.Globalorlocal, ResourceType)
+
+ Path := PONRMgr.GetPath(IntfID, ResourceType)
+ if Path == "" {
+ log.Errorf("Failed to get path for resource type %s", ResourceType)
+ return nil, fmt.Errorf("Failed to get path for resource type %s", ResourceType)
+ }
+ log.Debugf("Get resource for type %s on path %s", ResourceType, Path)
+ var Result []uint32
+ var NextID uint32
+ Resource, err := PONRMgr.GetResource(ctx, Path)
+ if (err == nil) && (ResourceType == ONU_ID) || (ResourceType == FLOW_ID) {
+ if NextID, err = PONRMgr.GenerateNextID(Resource); err != nil {
+ log.Error("Failed to Generate ID")
+ return Result, err
+ }
+ Result = append(Result, NextID)
+ } else if (err == nil) && ((ResourceType == GEMPORT_ID) || (ResourceType == ALLOC_ID)) {
+ if NumIDs == 1 {
+ if NextID, err = PONRMgr.GenerateNextID(Resource); err != nil {
+ log.Error("Failed to Generate ID")
+ return Result, err
+ }
+ Result = append(Result, NextID)
+ } else {
+ for NumIDs > 0 {
+ if NextID, err = PONRMgr.GenerateNextID(Resource); err != nil {
+ log.Error("Failed to Generate ID")
+ return Result, err
+ }
+ Result = append(Result, NextID)
+ NumIDs--
+ }
+ }
+ } else {
+ log.Error("get resource failed")
+ return Result, err
+ }
+
+ //Update resource in kv store
+ if PONRMgr.UpdateResource(ctx, Path, Resource) != nil {
+ log.Errorf("Failed to update resource %s", Path)
+ return nil, fmt.Errorf("Failed to update resource %s", Path)
+ }
+ return Result, nil
+}
+
+func checkValidResourceType(ResourceType string) bool {
+ KnownResourceTypes := []string{ONU_ID, ALLOC_ID, GEMPORT_ID, FLOW_ID}
+
+ for _, v := range KnownResourceTypes {
+ if v == ResourceType {
+ return true
+ }
+ }
+ return false
+}
+
+func (PONRMgr *PONResourceManager) FreeResourceID(ctx context.Context, IntfID uint32, ResourceType string, ReleaseContent []uint32) bool {
+ /*
+ Release alloc/gemport/onu/flow id for given OLT PON interface.
+ :param pon_intf_id: OLT PON interface id
+ :param resource_type: String to identify type of resource
+ :param release_content: required number of ids
+ :return boolean: True if all IDs in given release_content release else False
+ */
+ if !checkValidResourceType(ResourceType) {
+ log.Error("Invalid resource type")
+ return false
+ }
+ if ReleaseContent == nil {
+ log.Debug("Nothing to release")
+ return true
+ }
+ // delegate to the master instance if sharing enabled across instances
+ SharedResourceMgr := PONRMgr.SharedResourceMgrs[PONRMgr.SharedIdxByType[ResourceType]]
+ if SharedResourceMgr != nil && PONRMgr != SharedResourceMgr {
+ return SharedResourceMgr.FreeResourceID(ctx, IntfID, ResourceType, ReleaseContent)
+ }
+ Path := PONRMgr.GetPath(IntfID, ResourceType)
+ if Path == "" {
+ log.Error("Failed to get path")
+ return false
+ }
+ Resource, err := PONRMgr.GetResource(ctx, Path)
+ if err != nil {
+ log.Error("Failed to get resource")
+ return false
+ }
+ for _, Val := range ReleaseContent {
+ PONRMgr.ReleaseID(Resource, Val)
+ }
+ if PONRMgr.UpdateResource(ctx, Path, Resource) != nil {
+ log.Errorf("Free resource for %s failed", Path)
+ return false
+ }
+ return true
+}
+
+func (PONRMgr *PONResourceManager) UpdateResource(ctx context.Context, Path string, Resource map[string]interface{}) error {
+ /*
+ Update resource in resource kv store.
+ :param path: path to update resource
+ :param resource: resource need to be updated
+ :return boolean: True if resource updated in kv store else False
+ */
+ // TODO resource[POOL] = resource[POOL].bin
+ Value, err := json.Marshal(Resource)
+ if err != nil {
+ log.Error("failed to Marshal")
+ return err
+ }
+ err = PONRMgr.KVStore.Put(ctx, Path, Value)
+ if err != nil {
+ log.Error("failed to put data to kv store %s", Path)
+ return err
+ }
+ return nil
+}
+
+func (PONRMgr *PONResourceManager) ClearResourceIDPool(ctx context.Context, contIntfID uint32, ResourceType string) bool {
+ /*
+ Clear Resource Pool for a given Resource Type on a given PON Port.
+ :return boolean: True if removed else False
+ */
+
+ // delegate to the master instance if sharing enabled across instances
+ SharedResourceMgr := PONRMgr.SharedResourceMgrs[PONRMgr.SharedIdxByType[ResourceType]]
+ if SharedResourceMgr != nil && PONRMgr != SharedResourceMgr {
+ return SharedResourceMgr.ClearResourceIDPool(ctx, contIntfID, ResourceType)
+ }
+ Path := PONRMgr.GetPath(contIntfID, ResourceType)
+ if Path == "" {
+ log.Error("Failed to get path")
+ return false
+ }
+
+ if err := PONRMgr.KVStore.Delete(ctx, Path); err != nil {
+ log.Errorf("Failed to delete resource %s", Path)
+ return false
+ }
+ log.Debugf("Cleared resource %s", Path)
+ return true
+}
+
+func (PONRMgr PONResourceManager) InitResourceMap(ctx context.Context, PONIntfONUID string) {
+ /*
+ Initialize resource map
+ :param pon_intf_onu_id: reference of PON interface id and onu id
+ */
+ // initialize pon_intf_onu_id tuple to alloc_ids map
+ AllocIDPath := fmt.Sprintf(ALLOC_ID_RESOURCE_MAP_PATH, PONRMgr.DeviceID, PONIntfONUID)
+ var AllocIDs []byte
+ Result := PONRMgr.KVStore.Put(ctx, AllocIDPath, AllocIDs)
+ if Result != nil {
+ log.Error("Failed to update the KV store")
+ return
+ }
+ // initialize pon_intf_onu_id tuple to gemport_ids map
+ GEMPortIDPath := fmt.Sprintf(GEMPORT_ID_RESOURCE_MAP_PATH, PONRMgr.DeviceID, PONIntfONUID)
+ var GEMPortIDs []byte
+ Result = PONRMgr.KVStore.Put(ctx, GEMPortIDPath, GEMPortIDs)
+ if Result != nil {
+ log.Error("Failed to update the KV store")
+ return
+ }
+}
+
+func (PONRMgr PONResourceManager) RemoveResourceMap(ctx context.Context, PONIntfONUID string) bool {
+ /*
+ Remove resource map
+ :param pon_intf_onu_id: reference of PON interface id and onu id
+ */
+ // remove pon_intf_onu_id tuple to alloc_ids map
+ var err error
+ AllocIDPath := fmt.Sprintf(ALLOC_ID_RESOURCE_MAP_PATH, PONRMgr.DeviceID, PONIntfONUID)
+ if err = PONRMgr.KVStore.Delete(ctx, AllocIDPath); err != nil {
+ log.Errorf("Failed to remove resource %s", AllocIDPath)
+ return false
+ }
+ // remove pon_intf_onu_id tuple to gemport_ids map
+ GEMPortIDPath := fmt.Sprintf(GEMPORT_ID_RESOURCE_MAP_PATH, PONRMgr.DeviceID, PONIntfONUID)
+ err = PONRMgr.KVStore.Delete(ctx, GEMPortIDPath)
+ if err != nil {
+ log.Errorf("Failed to remove resource %s", GEMPortIDPath)
+ return false
+ }
+
+ FlowIDPath := fmt.Sprintf(FLOW_ID_RESOURCE_MAP_PATH, PONRMgr.DeviceID, PONIntfONUID)
+ if FlowIDs, err := PONRMgr.KVStore.List(ctx, FlowIDPath); err != nil {
+ for _, Flow := range FlowIDs {
+ FlowIDInfoPath := fmt.Sprintf(FLOW_ID_INFO_PATH, PONRMgr.DeviceID, PONIntfONUID, Flow.Value)
+ if err = PONRMgr.KVStore.Delete(ctx, FlowIDInfoPath); err != nil {
+ log.Errorf("Failed to remove resource %s", FlowIDInfoPath)
+ return false
+ }
+ }
+ }
+
+ if err = PONRMgr.KVStore.Delete(ctx, FlowIDPath); err != nil {
+ log.Errorf("Failed to remove resource %s", FlowIDPath)
+ return false
+ }
+
+ return true
+}
+
+func (PONRMgr *PONResourceManager) GetCurrentAllocIDForOnu(ctx context.Context, IntfONUID string) []uint32 {
+ /*
+ Get currently configured alloc ids for given pon_intf_onu_id
+ :param pon_intf_onu_id: reference of PON interface id and onu id
+ :return list: List of alloc_ids if available, else None
+ */
+ Path := fmt.Sprintf(ALLOC_ID_RESOURCE_MAP_PATH, PONRMgr.DeviceID, IntfONUID)
+
+ var Data []uint32
+ Value, err := PONRMgr.KVStore.Get(ctx, Path)
+ if err == nil {
+ if Value != nil {
+ Val, err := ToByte(Value.Value)
+ if err != nil {
+ log.Errorw("Failed to convert into byte array", log.Fields{"error": err})
+ return Data
+ }
+ if err = json.Unmarshal(Val, &Data); err != nil {
+ log.Error("Failed to unmarshal", log.Fields{"error": err})
+ return Data
+ }
+ }
+ }
+ return Data
+}
+
+func (PONRMgr *PONResourceManager) GetCurrentGEMPortIDsForOnu(ctx context.Context, IntfONUID string) []uint32 {
+ /*
+ Get currently configured gemport ids for given pon_intf_onu_id
+ :param pon_intf_onu_id: reference of PON interface id and onu id
+ :return list: List of gemport IDs if available, else None
+ */
+
+ Path := fmt.Sprintf(GEMPORT_ID_RESOURCE_MAP_PATH, PONRMgr.DeviceID, IntfONUID)
+ log.Debugf("Getting current gemports for %s", Path)
+ var Data []uint32
+ Value, err := PONRMgr.KVStore.Get(ctx, Path)
+ if err == nil {
+ if Value != nil {
+ Val, _ := ToByte(Value.Value)
+ if err = json.Unmarshal(Val, &Data); err != nil {
+ log.Errorw("Failed to unmarshal", log.Fields{"error": err})
+ return Data
+ }
+ }
+ } else {
+ log.Errorf("Failed to get data from kvstore for %s", Path)
+ }
+ return Data
+}
+
+func (PONRMgr *PONResourceManager) GetCurrentFlowIDsForOnu(ctx context.Context, IntfONUID string) []uint32 {
+ /*
+ Get currently configured flow ids for given pon_intf_onu_id
+ :param pon_intf_onu_id: reference of PON interface id and onu id
+ :return list: List of Flow IDs if available, else None
+ */
+
+ Path := fmt.Sprintf(FLOW_ID_RESOURCE_MAP_PATH, PONRMgr.DeviceID, IntfONUID)
+
+ var Data []uint32
+ Value, err := PONRMgr.KVStore.Get(ctx, Path)
+ if err == nil {
+ if Value != nil {
+ Val, _ := ToByte(Value.Value)
+ if err = json.Unmarshal(Val, &Data); err != nil {
+ log.Error("Failed to unmarshal")
+ return Data
+ }
+ }
+ }
+ return Data
+}
+
+func (PONRMgr *PONResourceManager) GetFlowIDInfo(ctx context.Context, IntfONUID string, FlowID uint32, Data interface{}) error {
+ /*
+ Get flow details configured for the ONU.
+ :param pon_intf_onu_id: reference of PON interface id and onu id
+ :param flow_id: Flow Id reference
+ :param Data: Result
+ :return error: nil if no error in getting from KV store
+ */
+
+ Path := fmt.Sprintf(FLOW_ID_INFO_PATH, PONRMgr.DeviceID, IntfONUID, FlowID)
+
+ Value, err := PONRMgr.KVStore.Get(ctx, Path)
+ if err == nil {
+ if Value != nil {
+ Val, err := ToByte(Value.Value)
+ if err != nil {
+ log.Errorw("Failed to convert flowinfo into byte array", log.Fields{"error": err})
+ return err
+ }
+ if err = json.Unmarshal(Val, Data); err != nil {
+ log.Errorw("Failed to unmarshal", log.Fields{"error": err})
+ return err
+ }
+ }
+ }
+ return err
+}
+
+func (PONRMgr *PONResourceManager) RemoveFlowIDInfo(ctx context.Context, IntfONUID string, FlowID uint32) bool {
+ /*
+ Get flow_id details configured for the ONU.
+ :param pon_intf_onu_id: reference of PON interface id and onu id
+ :param flow_id: Flow Id reference
+ */
+ Path := fmt.Sprintf(FLOW_ID_INFO_PATH, PONRMgr.DeviceID, IntfONUID, FlowID)
+
+ if err := PONRMgr.KVStore.Delete(ctx, Path); err != nil {
+ log.Errorf("Falied to remove resource %s", Path)
+ return false
+ }
+ return true
+}
+
+func (PONRMgr *PONResourceManager) UpdateAllocIdsForOnu(ctx context.Context, IntfONUID string, AllocIDs []uint32) error {
+ /*
+ Update currently configured alloc ids for given pon_intf_onu_id
+ :param pon_intf_onu_id: reference of PON interface id and onu id
+ :param alloc_ids: list of alloc ids
+ */
+ var Value []byte
+ var err error
+ Path := fmt.Sprintf(ALLOC_ID_RESOURCE_MAP_PATH, PONRMgr.DeviceID, IntfONUID)
+ Value, err = json.Marshal(AllocIDs)
+ if err != nil {
+ log.Error("failed to Marshal")
+ return err
+ }
+
+ if err = PONRMgr.KVStore.Put(ctx, Path, Value); err != nil {
+ log.Errorf("Failed to update resource %s", Path)
+ return err
+ }
+ return err
+}
+
+func (PONRMgr *PONResourceManager) UpdateGEMPortIDsForOnu(ctx context.Context, IntfONUID string, GEMPortIDs []uint32) error {
+ /*
+ Update currently configured gemport ids for given pon_intf_onu_id
+ :param pon_intf_onu_id: reference of PON interface id and onu id
+ :param gemport_ids: list of gem port ids
+ */
+
+ var Value []byte
+ var err error
+ Path := fmt.Sprintf(GEMPORT_ID_RESOURCE_MAP_PATH, PONRMgr.DeviceID, IntfONUID)
+ log.Debugf("Updating gemport ids for %s", Path)
+ Value, err = json.Marshal(GEMPortIDs)
+ if err != nil {
+ log.Error("failed to Marshal")
+ return err
+ }
+
+ if err = PONRMgr.KVStore.Put(ctx, Path, Value); err != nil {
+ log.Errorf("Failed to update resource %s", Path)
+ return err
+ }
+ return err
+}
+
+func checkForFlowIDInList(FlowIDList []uint32, FlowID uint32) (bool, uint32) {
+ /*
+ Check for a flow id in a given list of flow IDs.
+ :param FLowIDList: List of Flow IDs
+ :param FlowID: Flowd to check in the list
+ : return true and the index if present false otherwise.
+ */
+
+ for idx := range FlowIDList {
+ if FlowID == FlowIDList[idx] {
+ return true, uint32(idx)
+ }
+ }
+ return false, 0
+}
+
+func (PONRMgr *PONResourceManager) UpdateFlowIDForOnu(ctx context.Context, IntfONUID string, FlowID uint32, Add bool) error {
+ /*
+ Update the flow_id list of the ONU (add or remove flow_id from the list)
+ :param pon_intf_onu_id: reference of PON interface id and onu id
+ :param flow_id: flow ID
+ :param add: Boolean flag to indicate whether the flow_id should be
+ added or removed from the list. Defaults to adding the flow.
+ */
+ var Value []byte
+ var err error
+ var RetVal bool
+ var IDx uint32
+ Path := fmt.Sprintf(FLOW_ID_RESOURCE_MAP_PATH, PONRMgr.DeviceID, IntfONUID)
+ FlowIDs := PONRMgr.GetCurrentFlowIDsForOnu(ctx, IntfONUID)
+
+ if Add {
+ if RetVal, _ = checkForFlowIDInList(FlowIDs, FlowID); RetVal {
+ return nil
+ }
+ FlowIDs = append(FlowIDs, FlowID)
+ } else {
+ if RetVal, IDx = checkForFlowIDInList(FlowIDs, FlowID); !RetVal {
+ return nil
+ }
+ // delete the index and shift
+ FlowIDs = append(FlowIDs[:IDx], FlowIDs[IDx+1:]...)
+ }
+ Value, err = json.Marshal(FlowIDs)
+ if err != nil {
+ log.Error("Failed to Marshal")
+ return err
+ }
+
+ if err = PONRMgr.KVStore.Put(ctx, Path, Value); err != nil {
+ log.Errorf("Failed to update resource %s", Path)
+ return err
+ }
+ return err
+}
+
+func (PONRMgr *PONResourceManager) UpdateFlowIDInfoForOnu(ctx context.Context, IntfONUID string, FlowID uint32, FlowData interface{}) error {
+ /*
+ Update any metadata associated with the flow_id. The flow_data could be json
+ or any of other data structure. The resource manager doesnt care
+ :param pon_intf_onu_id: reference of PON interface id and onu id
+ :param flow_id: Flow ID
+ :param flow_data: Flow data blob
+ */
+ var Value []byte
+ var err error
+ Path := fmt.Sprintf(FLOW_ID_INFO_PATH, PONRMgr.DeviceID, IntfONUID, FlowID)
+ Value, err = json.Marshal(FlowData)
+ if err != nil {
+ log.Error("failed to Marshal")
+ return err
+ }
+
+ if err = PONRMgr.KVStore.Put(ctx, Path, Value); err != nil {
+ log.Errorf("Failed to update resource %s", Path)
+ return err
+ }
+ return err
+}
+
+func (PONRMgr *PONResourceManager) GenerateNextID(Resource map[string]interface{}) (uint32, error) {
+ /*
+ Generate unique id having OFFSET as start
+ :param resource: resource used to generate ID
+ :return uint32: generated id
+ */
+ ByteArray, err := ToByte(Resource[POOL])
+ if err != nil {
+ log.Error("Failed to convert resource to byte array")
+ return 0, err
+ }
+ Data := bitmap.TSFromData(ByteArray, false)
+ if Data == nil {
+ log.Error("Failed to get data from byte array")
+ return 0, errors.New("Failed to get data from byte array")
+ }
+
+ Len := Data.Len()
+ var Idx int
+ for Idx = 0; Idx < Len; Idx++ {
+ if !Data.Get(Idx) {
+ break
+ }
+ }
+ Data.Set(Idx, true)
+ res := uint32(Resource[START_IDX].(float64))
+ Resource[POOL] = Data.Data(false)
+ log.Debugf("Generated ID for %d", (uint32(Idx) + res))
+ return (uint32(Idx) + res), err
+}
+
+func (PONRMgr *PONResourceManager) ReleaseID(Resource map[string]interface{}, Id uint32) bool {
+ /*
+ Release unique id having OFFSET as start index.
+ :param resource: resource used to release ID
+ :param unique_id: id need to be released
+ */
+ ByteArray, err := ToByte(Resource[POOL])
+ if err != nil {
+ log.Error("Failed to convert resource to byte array")
+ return false
+ }
+ Data := bitmap.TSFromData(ByteArray, false)
+ if Data == nil {
+ log.Error("Failed to get resource pool")
+ return false
+ }
+ Idx := Id - uint32(Resource[START_IDX].(float64))
+ Data.Set(int(Idx), false)
+ Resource[POOL] = Data.Data(false)
+
+ return true
+}
+
+/* Reserves a unique id in the specified resource pool.
+:param Resource: resource used to reserve ID
+:param Id: ID to be reserved
+*/
+func (PONRMgr *PONResourceManager) reserveID(TSData *bitmap.Threadsafe, StartIndex uint32, Id uint32) bool {
+ Data := bitmap.TSFromData(TSData.Data(false), false)
+ if Data == nil {
+ log.Error("Failed to get resource pool")
+ return false
+ }
+ Idx := Id - StartIndex
+ Data.Set(int(Idx), true)
+ return true
+}
+
+func (PONRMgr *PONResourceManager) GetTechnology() string {
+ return PONRMgr.Technology
+}
+
+func (PONRMgr *PONResourceManager) GetResourceTypeAllocID() string {
+ return ALLOC_ID
+}
+
+func (PONRMgr *PONResourceManager) GetResourceTypeGemPortID() string {
+ return GEMPORT_ID
+}
+
+// ToByte converts an interface value to a []byte. The interface should either be of
+// a string type or []byte. Otherwise, an error is returned.
+func ToByte(value interface{}) ([]byte, error) {
+ switch t := value.(type) {
+ case []byte:
+ return value.([]byte), nil
+ case string:
+ return []byte(value.(string)), nil
+ default:
+ return nil, fmt.Errorf("unexpected-type-%T", t)
+ }
+}
+
+// ToString converts an interface value to a string. The interface should either be of
+// a string type or []byte. Otherwise, an error is returned.
+func ToString(value interface{}) (string, error) {
+ switch t := value.(type) {
+ case []byte:
+ return string(value.([]byte)), nil
+ case string:
+ return value.(string), nil
+ default:
+ return "", fmt.Errorf("unexpected-type-%T", t)
+ }
+}
+
+func (PONRMgr *PONResourceManager) AddOnuGemInfo(ctx context.Context, intfID uint32, onuGemData interface{}) error {
+ /*
+ Update onugem info map,
+ :param pon_intf_id: reference of PON interface id
+ :param onuegmdata: onugem info map
+ */
+ var Value []byte
+ var err error
+ Path := fmt.Sprintf(ONU_GEM_INFO_PATH, PONRMgr.DeviceID, intfID)
+ Value, err = json.Marshal(onuGemData)
+ if err != nil {
+ log.Error("failed to Marshal")
+ return err
+ }
+
+ if err = PONRMgr.KVStore.Put(ctx, Path, Value); err != nil {
+ log.Errorf("Failed to update resource %s", Path)
+ return err
+ }
+ return err
+}
+
+func (PONRMgr *PONResourceManager) GetOnuGemInfo(ctx context.Context, IntfId uint32, onuGemInfo interface{}) error {
+ /*
+ Get onugeminfo map from kvstore
+ :param intfid: refremce pon intfid
+ :param onuGemInfo: onugem info to return from kv strore.
+ */
+ var Val []byte
+
+ path := fmt.Sprintf(ONU_GEM_INFO_PATH, PONRMgr.DeviceID, IntfId)
+ value, err := PONRMgr.KVStore.Get(ctx, path)
+ if err != nil {
+ log.Errorw("Failed to get from kv store", log.Fields{"path": path})
+ return err
+ } else if value == nil {
+ log.Debug("No onuinfo for path", log.Fields{"path": path})
+ return nil // returning nil as this could happen if there are no onus for the interface yet
+ }
+ if Val, err = kvstore.ToByte(value.Value); err != nil {
+ log.Error("Failed to convert to byte array")
+ return err
+ }
+
+ if err = json.Unmarshal(Val, &onuGemInfo); err != nil {
+ log.Error("Failed to unmarshall")
+ return err
+ }
+ log.Debugw("found onuinfo from path", log.Fields{"path": path, "onuinfo": onuGemInfo})
+ return err
+}
+
+func (PONRMgr *PONResourceManager) DelOnuGemInfoForIntf(ctx context.Context, intfId uint32) error {
+ /*
+ delete onugem info for an interface from kvstore
+ :param intfid: refremce pon intfid
+ */
+
+ path := fmt.Sprintf(ONU_GEM_INFO_PATH, PONRMgr.DeviceID, intfId)
+ if err := PONRMgr.KVStore.Delete(ctx, path); err != nil {
+ log.Errorf("Falied to remove resource %s", path)
+ return err
+ }
+ return nil
+}
diff --git a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/techprofile/4QueueHybridProfileMap1.json b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/techprofile/4QueueHybridProfileMap1.json
new file mode 100644
index 0000000..d11f8e4
--- /dev/null
+++ b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/techprofile/4QueueHybridProfileMap1.json
@@ -0,0 +1,141 @@
+ {
+ "name": "4QueueHybridProfileMap1",
+ "profile_type": "XPON",
+ "version": 1,
+ "num_gem_ports": 4,
+ "instance_control": {
+ "onu": "multi-instance",
+ "uni": "single-instance",
+ "max_gem_payload_size": "auto"
+ },
+ "us_scheduler": {
+ "additional_bw": "AdditionalBW_Auto",
+ "direction": "UPSTREAM",
+ "priority": 0,
+ "weight": 0,
+ "q_sched_policy": "Hybrid"
+ },
+ "ds_scheduler": {
+ "additional_bw": "AdditionalBW_Auto",
+ "direction": "DOWNSTREAM",
+ "priority": 0,
+ "weight": 0,
+ "q_sched_policy": "Hybrid"
+ },
+ "upstream_gem_port_attribute_list": [
+ {
+ "pbit_map": "0b00000101",
+ "aes_encryption": "True",
+ "scheduling_policy": "WRR",
+ "priority_q": 4,
+ "weight": 25,
+ "discard_policy": "TailDrop",
+ "max_q_size": "auto",
+ "discard_config": {
+ "max_threshold": 0,
+ "min_threshold": 0,
+ "max_probability": 0
+ }
+ },
+ {
+ "pbit_map": "0b00011010",
+ "aes_encryption": "True",
+ "scheduling_policy": "WRR",
+ "priority_q": 3,
+ "weight": 75,
+ "discard_policy": "TailDrop",
+ "max_q_size": "auto",
+ "discard_config": {
+ "min_threshold": 0,
+ "max_threshold": 0,
+ "max_probability": 0
+ }
+ },
+ {
+ "pbit_map": "0b00100000",
+ "aes_encryption": "True",
+ "scheduling_policy": "StrictPriority",
+ "priority_q": 2,
+ "weight": 0,
+ "discard_policy": "TailDrop",
+ "max_q_size": "auto",
+ "discard_config": {
+ "min_threshold": 0,
+ "max_threshold": 0,
+ "max_probability": 0
+ }
+ },
+ {
+ "pbit_map": "0b11000000",
+ "aes_encryption": "True",
+ "scheduling_policy": "StrictPriority",
+ "priority_q": 1,
+ "weight": 25,
+ "discard_policy": "TailDrop",
+ "max_q_size": "auto",
+ "discard_config": {
+ "min_threshold": 0,
+ "max_threshold": 0,
+ "max_probability": 0
+ }
+ }
+ ],
+ "downstream_gem_port_attribute_list": [
+ {
+ "pbit_map": "0b00000101",
+ "aes_encryption": "True",
+ "scheduling_policy": "WRR",
+ "priority_q": 4,
+ "weight": 10,
+ "discard_policy": "TailDrop",
+ "max_q_size": "auto",
+ "discard_config": {
+ "min_threshold": 0,
+ "max_threshold": 0,
+ "max_probability": 0
+ }
+ },
+ {
+ "pbit_map": "0b00011010",
+ "aes_encryption": "True",
+ "scheduling_policy": "WRR",
+ "priority_q": 3,
+ "weight": 90,
+ "discard_policy": "TailDrop",
+ "max_q_size": "auto",
+ "discard_config": {
+ "min_threshold": 0,
+ "max_threshold": 0,
+ "max_probability": 0
+ }
+ },
+ {
+ "pbit_map": "0b00100000",
+ "aes_encryption": "True",
+ "scheduling_policy": "StrictPriority",
+ "priority_q": 2,
+ "weight": 0,
+ "discard_policy": "TailDrop",
+ "max_q_size": "auto",
+ "discard_config": {
+ "min_threshold": 0,
+ "max_threshold": 0,
+ "max_probability": 0
+ }
+ },
+ {
+ "pbit_map": "0b11000000",
+ "aes_encryption": "True",
+ "scheduling_policy": "StrictPriority",
+ "priority_q": 1,
+ "weight": 25,
+ "discard_policy": "TailDrop",
+ "max_q_size": "auto",
+ "discard_config": {
+ "min_threshold": 0,
+ "max_threshold": 0,
+ "max_probability": 0
+ }
+ }
+ ]
+}
diff --git a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/techprofile/README.md b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/techprofile/README.md
new file mode 100644
index 0000000..03a396d
--- /dev/null
+++ b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/techprofile/README.md
@@ -0,0 +1,336 @@
+Technology Profile Management
+Overview
+Technology profiles that are utilized by VOLTHA are stored in a prescribed structure in VOLTHA's key/value store, which is currently etcd. The key structure used to access technology profiles is /voltha/technology_profiles//; where TID is the numeric ID of the technology profile and TECHNOLOGY specifies the technology being utilized by the adapter, e.g. xgspon. While the TID key is a directory, the TECHNOLOGY key should be set to the JSON data that represents the technology profile values.
+
+NOTE: The content of a technology profile represents a contract between the technology profile definition and all adapters that consume that technology profile. The structure and content of the profiles are outside the scope of Technology Profile Management. Technology profile management only specifies the key/value structure in which profiles are stored.
+
+Example JSON :
+
+{
+ "name": "4QueueHybridProfileMap1",
+ "profile_type": "XPON",
+ "version": 1,
+ "num_gem_ports": 4,
+ "instance_control": {
+ "onu": "multi-instance",
+ "uni": "single-instance",
+ "max_gem_payload_size": "auto"
+ },
+ "us_scheduler": {
+ "additional_bw": "auto",
+ "direction": "UPSTREAM",
+ "priority": 0,
+ "weight": 0,
+ "q_sched_policy": "hybrid"
+ },
+ "ds_scheduler": {
+ "additional_bw": "auto",
+ "direction": "DOWNSTREAM",
+ "priority": 0,
+ "weight": 0,
+ "q_sched_policy": "hybrid"
+ },
+ "upstream_gem_port_attribute_list": [
+ {
+ "pbit_map": "0b00000101",
+ "aes_encryption": "True",
+ "scheduling_policy": "WRR",
+ "priority_q": 4,
+ "weight": 25,
+ "discard_policy": "TailDrop",
+ "max_q_size": "auto",
+ "discard_config": {
+ "max_threshold": 0,
+ "min_threshold": 0,
+ "max_probability": 0
+ }
+ },
+ {
+ "pbit_map": "0b00011010",
+ "aes_encryption": "True",
+ "scheduling_policy": "WRR",
+ "priority_q": 3,
+ "weight": 75,
+ "discard_policy": "TailDrop",
+ "max_q_size": "auto",
+ "discard_config": {
+ "min_threshold": 0,
+ "max_threshold": 0,
+ "max_probability": 0
+ }
+ },
+ {
+ "pbit_map": "0b00100000",
+ "aes_encryption": "True",
+ "scheduling_policy": "StrictPriority",
+ "priority_q": 2,
+ "weight": 0,
+ "discard_policy": "TailDrop",
+ "max_q_size": "auto",
+ "discard_config": {
+ "min_threshold": 0,
+ "max_threshold": 0,
+ "max_probability": 0
+ }
+ },
+ {
+ "pbit_map": "0b11000000",
+ "aes_encryption": "True",
+ "scheduling_policy": "StrictPriority",
+ "priority_q": 1,
+ "weight": 25,
+ "discard_policy": "TailDrop",
+ "max_q_size": "auto",
+ "discard_config": {
+ "min_threshold": 0,
+ "max_threshold": 0,
+ "max_probability": 0
+ }
+ }
+ ],
+ "downstream_gem_port_attribute_list": [
+ {
+ "pbit_map": "0b00000101",
+ "aes_encryption": "True",
+ "scheduling_policy": "WRR",
+ "priority_q": 4,
+ "weight": 10,
+ "discard_policy": "TailDrop",
+ "max_q_size": "auto",
+ "discard_config": {
+ "min_threshold": 0,
+ "max_threshold": 0,
+ "max_probability": 0
+ }
+ },
+ {
+ "pbit_map": "0b00011010",
+ "aes_encryption": "True",
+ "scheduling_policy": "WRR",
+ "priority_q": 3,
+ "weight": 90,
+ "discard_policy": "TailDrop",
+ "max_q_size": "auto",
+ "discard_config": {
+ "min_threshold": 0,
+ "max_threshold": 0,
+ "max_probability": 0
+ }
+ },
+ {
+ "pbit_map": "0b00100000",
+ "aes_encryption": "True",
+ "scheduling_policy": "StrictPriority",
+ "priority_q": 2,
+ "weight": 0,
+ "discard_policy": "TailDrop",
+ "max_q_size": "auto",
+ "discard_config": {
+ "min_threshold": 0,
+ "max_threshold": 0,
+ "max_probability": 0
+ }
+ },
+ {
+ "pbit_map": "0b11000000",
+ "aes_encryption": "True",
+ "scheduling_policy": "StrictPriority",
+ "priority_q": 1,
+ "weight": 25,
+ "discard_policy": "TailDrop",
+ "max_q_size": "auto",
+ "discard_config": {
+ "min_threshold": 0,
+ "max_threshold": 0,
+ "max_probability": 0
+ }
+ }
+ ]
+}
+
+Creating Technology Profiles
+Technology profiles are a simple JSON object. This JSON object can be created using a variety of tools such as Vim, Emacs, or various IDEs. JQ can be a useful tool for validating a JSON object. Once a file is created with the JSON object it can be stored in VOLTHA key/value store using the standard etcd command line tool etcdctl or using an HTTP POST operation using Curl.
+
+Assuming you are in a standard VOLTHA deployment within a Kubernetes cluster you can access the etcd key/value store using kubectl via the PODs named etcd-cluster-0000, etcd-cluster-0001, or etcd-cluster-0002. For the examples in this document etcd-cluster-0000 will be used, but it really shouldn't matter which is used.
+
+ETCD version 3 is being used in techprofile module : Export this variable before using curl operation , export ETCDCTL_API=3
+
+Assuming the Technology template is stored in a local file 4QueueHybridProfileMap1.json the following commands could be used to store or update the technical template into the proper location in the etcd key/value store:
+
+# Store a Technology template using etcdctl
+jq -c . 4QueueHybridProfileMap1.json | kubectl exec -i etcd-cluster-0000 -- etcdctl set service/voltha/technology_profiles/xgspon/64
+
+jq -c . 4QueueHybridProfileMap1.json | etcdctl --endpoints=<ETCDIP>:2379 put service/voltha/technology_profiles/xgspon/64
+
+
+# Store a Technology template using curl
+curl -sSL -XPUT http://10.233.53.161:2379/v2/keys/service/voltha/technology_profiles/xgspon/64 -d value="$(jq -c . 4QueueHybridProfileMap1.json)"
+In the examples above, the command jq is used. This command can be installed using standard package management tools on most Linux systems. In the examples the "-c" option is used to compress the JSON. Using this tool is not necessary, and if you choose not to use the tool, you can replace "jq -c ," in the above examples with the "cat" command. More on jq can be found at https://stedolan.github.io/jq/.
+
+Listing Technical Profiles for a given Technology
+While both curl and etcdctl (via kubectl) can be used to list or view the available Technology profiles, etcdctl is easier, and thus will be used in the examples. For listing Technology profiles etcdctl ls is used. In can be used in conjunction with the -r option to recursively list profiles.
+
+
+#List Tech profile
+etcdctl --endpoints=<EtcdIPAddres>:2379 get service/voltha/technology_profiles/xgspon/64
+
+
+# Example output
+A specified Technology profile can be viewed with the etcdctl get command. (Again, jq is used for presentation purposes, and is not required)
+
+# Display a specified Technology profile, using jq to pretty print
+kubectl exec -i etcd-cluster-0000 -- etcdctl get /xgspon/64 | jq .
+
+etcdctl --endpoints=<ETCDIP>:2379 get service/voltha/technology_profiles/xgspon/64
+# Example outpout
+service/voltha/technology_profiles/xgspon/64/uni-1
+{
+ "name": "4QueueHybridProfileMap1",
+ "profile_type": "XPON",
+ "version": 1,
+ "num_gem_ports": 4,
+ "instance_control": {
+ "onu": "multi-instance",
+ "uni": "single-instance",
+ "max_gem_payload_size": "auto"
+ },
+ "us_scheduler": {
+ "additional_bw": "auto",
+ "direction": "UPSTREAM",
+ "priority": 0,
+ "weight": 0,
+ "q_sched_policy": "hybrid"
+ },
+ "ds_scheduler": {
+ "additional_bw": "auto",
+ "direction": "DOWNSTREAM",
+ "priority": 0,
+ "weight": 0,
+ "q_sched_policy": "hybrid"
+ },
+ "upstream_gem_port_attribute_list": [
+ {
+ "pbit_map": "0b00000101",
+ "aes_encryption": "True",
+ "scheduling_policy": "WRR",
+ "priority_q": 4,
+ "weight": 25,
+ "discard_policy": "TailDrop",
+ "max_q_size": "auto",
+ "discard_config": {
+ "max_threshold": 0,
+ "min_threshold": 0,
+ "max_probability": 0
+ }
+ },
+ {
+ "pbit_map": "0b00011010",
+ "aes_encryption": "True",
+ "scheduling_policy": "WRR",
+ "priority_q": 3,
+ "weight": 75,
+ "discard_policy": "TailDrop",
+ "max_q_size": "auto",
+ "discard_config": {
+ "min_threshold": 0,
+ "max_threshold": 0,
+ "max_probability": 0
+ }
+ },
+ {
+ "pbit_map": "0b00100000",
+ "aes_encryption": "True",
+ "scheduling_policy": "StrictPriority",
+ "priority_q": 2,
+ "weight": 0,
+ "discard_policy": "TailDrop",
+ "max_q_size": "auto",
+ "discard_config": {
+ "min_threshold": 0,
+ "max_threshold": 0,
+ "max_probability": 0
+ }
+ },
+ {
+ "pbit_map": "0b11000000",
+ "aes_encryption": "True",
+ "scheduling_policy": "StrictPriority",
+ "priority_q": 1,
+ "weight": 25,
+ "discard_policy": "TailDrop",
+ "max_q_size": "auto",
+ "discard_config": {
+ "min_threshold": 0,
+ "max_threshold": 0,
+ "max_probability": 0
+ }
+ }
+ ],
+ "downstream_gem_port_attribute_list": [
+ {
+ "pbit_map": "0b00000101",
+ "aes_encryption": "True",
+ "scheduling_policy": "WRR",
+ "priority_q": 4,
+ "weight": 10,
+ "discard_policy": "TailDrop",
+ "max_q_size": "auto",
+ "discard_config": {
+ "min_threshold": 0,
+ "max_threshold": 0,
+ "max_probability": 0
+ }
+ },
+ {
+ "pbit_map": "0b00011010",
+ "aes_encryption": "True",
+ "scheduling_policy": "WRR",
+ "priority_q": 3,
+ "weight": 90,
+ "discard_policy": "TailDrop",
+ "max_q_size": "auto",
+ "discard_config": {
+ "min_threshold": 0,
+ "max_threshold": 0,
+ "max_probability": 0
+ }
+ },
+ {
+ "pbit_map": "0b00100000",
+ "aes_encryption": "True",
+ "scheduling_policy": "StrictPriority",
+ "priority_q": 2,
+ "weight": 0,
+ "discard_policy": "TailDrop",
+ "max_q_size": "auto",
+ "discard_config": {
+ "min_threshold": 0,
+ "max_threshold": 0,
+ "max_probability": 0
+ }
+ },
+ {
+ "pbit_map": "0b11000000",
+ "aes_encryption": "True",
+ "scheduling_policy": "StrictPriority",
+ "priority_q": 1,
+ "weight": 25,
+ "discard_policy": "TailDrop",
+ "max_q_size": "auto",
+ "discard_config": {
+ "min_threshold": 0,
+ "max_threshold": 0,
+ "max_probability": 0
+ }
+ }
+ ]
+}
+
+#Deleting Technology Profiles
+A technology profile or a technology profile tree can be removed using etcdctl rm.
+
+# Remove a specific technology profile
+kubectl exec -i etcd-cluster-0000 -- etcdctl rm /xgspon/64
+
+# Remove all technology profiles associated with Technology xgspon and ID 64(including the profile ID key)
+kubectl exec -i etcd-cluster-0000 -- etcdctl rm --dir -r /xgspon/64
diff --git a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/techprofile/config.go b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/techprofile/config.go
new file mode 100644
index 0000000..4af2bd5
--- /dev/null
+++ b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/techprofile/config.go
@@ -0,0 +1,115 @@
+/*
+ * Copyright 2019-present Open Networking Foundation
+
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+
+ * http://www.apache.org/licenses/LICENSE-2.0
+
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package techprofile
+
+import (
+ "github.com/opencord/voltha-lib-go/v3/pkg/db"
+)
+
+// tech profile default constants
+const (
+ defaultTechProfileName = "Default_1tcont_1gem_Profile"
+ DEFAULT_TECH_PROFILE_TABLE_ID = 64
+ defaultVersion = 1.0
+ defaultLogLevel = 0
+ defaultGemportsCount = 1
+ defaultPbits = "0b11111111"
+
+ defaultKVStoreTimeout = 5 //in seconds
+
+ // Tech profile path prefix in kv store
+ defaultKVPathPrefix = "service/voltha/technology_profiles"
+
+ // Tech profile path in kv store
+ defaultTechProfileKVPath = "%s/%d" // <technology>/<tech_profile_tableID>
+
+ // Tech profile instance path in kv store
+ // Format: <technology>/<tech_profile_tableID>/<uni_port_name>
+ defaultTPInstanceKVPath = "%s/%d/%s"
+)
+
+//Tech-Profile JSON String Keys
+// NOTE: Tech profile templeate JSON file should comply with below keys
+const (
+ NAME = "name"
+ PROFILE_TYPE = "profile_type"
+ VERSION = "version"
+ NUM_GEM_PORTS = "num_gem_ports"
+ INSTANCE_CONTROL = "instance_control"
+ US_SCHEDULER = "us_scheduler"
+ DS_SCHEDULER = "ds_scheduler"
+ UPSTREAM_GEM_PORT_ATTRIBUTE_LIST = "upstream_gem_port_attribute_list"
+ DOWNSTREAM_GEM_PORT_ATTRIBUTE_LIST = "downstream_gem_port_attribute_list"
+ ONU = "onu"
+ UNI = "uni"
+ MAX_GEM_PAYLOAD_SIZE = "max_gem_payload_size"
+ DIRECTION = "direction"
+ ADDITIONAL_BW = "additional_bw"
+ PRIORITY = "priority"
+ Q_SCHED_POLICY = "q_sched_policy"
+ WEIGHT = "weight"
+ PBIT_MAP = "pbit_map"
+ DISCARD_CONFIG = "discard_config"
+ MAX_THRESHOLD = "max_threshold"
+ MIN_THRESHOLD = "min_threshold"
+ MAX_PROBABILITY = "max_probability"
+ DISCARD_POLICY = "discard_policy"
+ PRIORITY_Q = "priority_q"
+ SCHEDULING_POLICY = "scheduling_policy"
+ MAX_Q_SIZE = "max_q_size"
+ AES_ENCRYPTION = "aes_encryption"
+)
+
+// TechprofileFlags represents the set of configurations used
+type TechProfileFlags struct {
+ KVStoreHost string
+ KVStorePort int
+ KVStoreType string
+ KVStoreTimeout int
+ KVBackend *db.Backend
+ TPKVPathPrefix string
+ TPFileKVPath string
+ TPInstanceKVPath string
+ DefaultTPName string
+ TPVersion int
+ NumGemPorts uint32
+ DefaultPbits []string
+ LogLevel int
+ DefaultTechProfileID uint32
+ DefaultNumGemPorts uint32
+}
+
+func NewTechProfileFlags(KVStoreType string, KVStoreHost string, KVStorePort int) *TechProfileFlags {
+ // initialize with default values
+ var techProfileFlags = TechProfileFlags{
+ KVBackend: nil,
+ KVStoreHost: KVStoreHost,
+ KVStorePort: KVStorePort,
+ KVStoreType: KVStoreType,
+ KVStoreTimeout: defaultKVStoreTimeout,
+ DefaultTPName: defaultTechProfileName,
+ TPKVPathPrefix: defaultKVPathPrefix,
+ TPVersion: defaultVersion,
+ TPFileKVPath: defaultTechProfileKVPath,
+ TPInstanceKVPath: defaultTPInstanceKVPath,
+ DefaultTechProfileID: DEFAULT_TECH_PROFILE_TABLE_ID,
+ DefaultNumGemPorts: defaultGemportsCount,
+ DefaultPbits: []string{defaultPbits},
+ LogLevel: defaultLogLevel,
+ }
+
+ return &techProfileFlags
+}
diff --git a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/techprofile/tech_profile.go b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/techprofile/tech_profile.go
new file mode 100644
index 0000000..ba8855f
--- /dev/null
+++ b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/techprofile/tech_profile.go
@@ -0,0 +1,923 @@
+/*
+ * Copyright 2019-present Open Networking Foundation
+
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+
+ * http://www.apache.org/licenses/LICENSE-2.0
+
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package techprofile
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "regexp"
+ "strconv"
+
+ "github.com/opencord/voltha-lib-go/v3/pkg/db"
+
+ "github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore"
+ "github.com/opencord/voltha-lib-go/v3/pkg/log"
+ tp_pb "github.com/opencord/voltha-protos/v3/go/tech_profile"
+)
+
+// Interface to pon resource manager APIs
+type iPonResourceMgr interface {
+ GetResourceID(ctx context.Context, IntfID uint32, ResourceType string, NumIDs uint32) ([]uint32, error)
+ GetResourceTypeAllocID() string
+ GetResourceTypeGemPortID() string
+ GetTechnology() string
+}
+
+type Direction int32
+
+const (
+ Direction_UPSTREAM Direction = 0
+ Direction_DOWNSTREAM Direction = 1
+ Direction_BIDIRECTIONAL Direction = 2
+)
+
+var Direction_name = map[Direction]string{
+ 0: "UPSTREAM",
+ 1: "DOWNSTREAM",
+ 2: "BIDIRECTIONAL",
+}
+
+type SchedulingPolicy int32
+
+const (
+ SchedulingPolicy_WRR SchedulingPolicy = 0
+ SchedulingPolicy_StrictPriority SchedulingPolicy = 1
+ SchedulingPolicy_Hybrid SchedulingPolicy = 2
+)
+
+var SchedulingPolicy_name = map[SchedulingPolicy]string{
+ 0: "WRR",
+ 1: "StrictPriority",
+ 2: "Hybrid",
+}
+
+type AdditionalBW int32
+
+const (
+ AdditionalBW_AdditionalBW_None AdditionalBW = 0
+ AdditionalBW_AdditionalBW_NA AdditionalBW = 1
+ AdditionalBW_AdditionalBW_BestEffort AdditionalBW = 2
+ AdditionalBW_AdditionalBW_Auto AdditionalBW = 3
+)
+
+var AdditionalBW_name = map[AdditionalBW]string{
+ 0: "AdditionalBW_None",
+ 1: "AdditionalBW_NA",
+ 2: "AdditionalBW_BestEffort",
+ 3: "AdditionalBW_Auto",
+}
+
+type DiscardPolicy int32
+
+const (
+ DiscardPolicy_TailDrop DiscardPolicy = 0
+ DiscardPolicy_WTailDrop DiscardPolicy = 1
+ DiscardPolicy_Red DiscardPolicy = 2
+ DiscardPolicy_WRed DiscardPolicy = 3
+)
+
+var DiscardPolicy_name = map[DiscardPolicy]string{
+ 0: "TailDrop",
+ 1: "WTailDrop",
+ 2: "Red",
+ 3: "WRed",
+}
+
+// Required uniPortName format
+var uniPortNameFormat = regexp.MustCompile(`^pon-{[0-9]+}/onu-{[0-9]+}/uni-{[0-9]+}$`)
+
+/*
+type InferredAdditionBWIndication int32
+
+const (
+ InferredAdditionBWIndication_InferredAdditionBWIndication_None InferredAdditionBWIndication = 0
+ InferredAdditionBWIndication_InferredAdditionBWIndication_Assured InferredAdditionBWIndication = 1
+ InferredAdditionBWIndication_InferredAdditionBWIndication_BestEffort InferredAdditionBWIndication = 2
+)
+
+var InferredAdditionBWIndication_name = map[int32]string{
+ 0: "InferredAdditionBWIndication_None",
+ 1: "InferredAdditionBWIndication_Assured",
+ 2: "InferredAdditionBWIndication_BestEffort",
+}
+*/
+// instance control defaults
+const (
+ defaultOnuInstance = "multi-instance"
+ defaultUniInstance = "single-instance"
+ defaultGemPayloadSize = "auto"
+)
+
+const MAX_GEM_PAYLOAD = "max_gem_payload_size"
+
+type InstanceControl struct {
+ Onu string `json:"ONU"`
+ Uni string `json:"uni"`
+ MaxGemPayloadSize string `json:"max_gem_payload_size"`
+}
+
+// default discard config constants
+const (
+ defaultMinThreshold = 0
+ defaultMaxThreshold = 0
+ defaultMaxProbability = 0
+)
+
+type DiscardConfig struct {
+ MinThreshold int `json:"min_threshold"`
+ MaxThreshold int `json:"max_threshold"`
+ MaxProbability int `json:"max_probability"`
+}
+
+// default scheduler contants
+const (
+ defaultAdditionalBw = AdditionalBW_AdditionalBW_BestEffort
+ defaultPriority = 0
+ defaultWeight = 0
+ defaultQueueSchedPolicy = SchedulingPolicy_Hybrid
+)
+
+type Scheduler struct {
+ Direction string `json:"direction"`
+ AdditionalBw string `json:"additional_bw"`
+ Priority uint32 `json:"priority"`
+ Weight uint32 `json:"weight"`
+ QSchedPolicy string `json:"q_sched_policy"`
+}
+
+// default GEM attribute constants
+const (
+ defaultAESEncryption = "True"
+ defaultPriorityQueue = 0
+ defaultQueueWeight = 0
+ defaultMaxQueueSize = "auto"
+ defaultdropPolicy = DiscardPolicy_TailDrop
+ defaultSchedulePolicy = SchedulingPolicy_WRR
+ defaultIsMulticast = "False"
+ defaultAccessControlList = "224.0.0.0-239.255.255.255"
+ defaultMcastGemID = 4069
+)
+
+type GemPortAttribute struct {
+ MaxQueueSize string `json:"max_q_size"`
+ PbitMap string `json:"pbit_map"`
+ AesEncryption string `json:"aes_encryption"`
+ SchedulingPolicy string `json:"scheduling_policy"`
+ PriorityQueue uint32 `json:"priority_q"`
+ Weight uint32 `json:"weight"`
+ DiscardPolicy string `json:"discard_policy"`
+ DiscardConfig DiscardConfig `json:"discard_config"`
+ IsMulticast string `json:"is_multicast"`
+ DControlList string `json:"dynamic_access_control_list"`
+ SControlList string `json:"static_access_control_list"`
+ McastGemID uint32 `json:"multicast_gem_id"`
+}
+
+type iScheduler struct {
+ AllocID uint32 `json:"alloc_id"`
+ Direction string `json:"direction"`
+ AdditionalBw string `json:"additional_bw"`
+ Priority uint32 `json:"priority"`
+ Weight uint32 `json:"weight"`
+ QSchedPolicy string `json:"q_sched_policy"`
+}
+type iGemPortAttribute struct {
+ GemportID uint32 `json:"gemport_id"`
+ MaxQueueSize string `json:"max_q_size"`
+ PbitMap string `json:"pbit_map"`
+ AesEncryption string `json:"aes_encryption"`
+ SchedulingPolicy string `json:"scheduling_policy"`
+ PriorityQueue uint32 `json:"priority_q"`
+ Weight uint32 `json:"weight"`
+ DiscardPolicy string `json:"discard_policy"`
+ DiscardConfig DiscardConfig `json:"discard_config"`
+ IsMulticast string `json:"is_multicast"`
+ DControlList string `json:"dynamic_access_control_list"`
+ SControlList string `json:"static_access_control_list"`
+ McastGemID uint32 `json:"multicast_gem_id"`
+}
+
+type TechProfileMgr struct {
+ config *TechProfileFlags
+ resourceMgr iPonResourceMgr
+}
+type DefaultTechProfile struct {
+ Name string `json:"name"`
+ ProfileType string `json:"profile_type"`
+ Version int `json:"version"`
+ NumGemPorts uint32 `json:"num_gem_ports"`
+ InstanceCtrl InstanceControl `json:"instance_control"`
+ UsScheduler Scheduler `json:"us_scheduler"`
+ DsScheduler Scheduler `json:"ds_scheduler"`
+ UpstreamGemPortAttributeList []GemPortAttribute `json:"upstream_gem_port_attribute_list"`
+ DownstreamGemPortAttributeList []GemPortAttribute `json:"downstream_gem_port_attribute_list"`
+}
+type TechProfile struct {
+ Name string `json:"name"`
+ SubscriberIdentifier string `json:"subscriber_identifier"`
+ ProfileType string `json:"profile_type"`
+ Version int `json:"version"`
+ NumGemPorts uint32 `json:"num_gem_ports"`
+ InstanceCtrl InstanceControl `json:"instance_control"`
+ UsScheduler iScheduler `json:"us_scheduler"`
+ DsScheduler iScheduler `json:"ds_scheduler"`
+ UpstreamGemPortAttributeList []iGemPortAttribute `json:"upstream_gem_port_attribute_list"`
+ DownstreamGemPortAttributeList []iGemPortAttribute `json:"downstream_gem_port_attribute_list"`
+}
+
+func (t *TechProfileMgr) SetKVClient() *db.Backend {
+ addr := t.config.KVStoreHost + ":" + strconv.Itoa(t.config.KVStorePort)
+ kvClient, err := newKVClient(t.config.KVStoreType, addr, t.config.KVStoreTimeout)
+ if err != nil {
+ log.Errorw("failed-to-create-kv-client",
+ log.Fields{
+ "type": t.config.KVStoreType, "host": t.config.KVStoreHost, "port": t.config.KVStorePort,
+ "timeout": t.config.KVStoreTimeout, "prefix": t.config.TPKVPathPrefix,
+ "error": err.Error(),
+ })
+ return nil
+ }
+ return &db.Backend{
+ Client: kvClient,
+ StoreType: t.config.KVStoreType,
+ Host: t.config.KVStoreHost,
+ Port: t.config.KVStorePort,
+ Timeout: t.config.KVStoreTimeout,
+ PathPrefix: t.config.TPKVPathPrefix}
+
+ /* TODO : Make sure direct call to NewBackend is working fine with backend , currently there is some
+ issue between kv store and backend , core is not calling NewBackend directly
+ kv := model.NewBackend(t.config.KVStoreType, t.config.KVStoreHost, t.config.KVStorePort,
+ t.config.KVStoreTimeout, kvStoreTechProfilePathPrefix)
+ */
+}
+
+func newKVClient(storeType string, address string, timeout int) (kvstore.Client, error) {
+
+ log.Infow("kv-store", log.Fields{"storeType": storeType, "address": address})
+ switch storeType {
+ case "consul":
+ return kvstore.NewConsulClient(address, timeout)
+ case "etcd":
+ return kvstore.NewEtcdClient(address, timeout)
+ }
+ return nil, errors.New("unsupported-kv-store")
+}
+
+func NewTechProfile(resourceMgr iPonResourceMgr, KVStoreType string, KVStoreHost string, KVStorePort int) (*TechProfileMgr, error) {
+ var techprofileObj TechProfileMgr
+ log.Debug("Initializing techprofile Manager")
+ techprofileObj.config = NewTechProfileFlags(KVStoreType, KVStoreHost, KVStorePort)
+ techprofileObj.config.KVBackend = techprofileObj.SetKVClient()
+ if techprofileObj.config.KVBackend == nil {
+ log.Error("Failed to initialize KV backend\n")
+ return nil, errors.New("KV backend init failed")
+ }
+ techprofileObj.resourceMgr = resourceMgr
+ log.Debug("Initializing techprofile object instance success")
+ return &techprofileObj, nil
+}
+
+func (t *TechProfileMgr) GetTechProfileInstanceKVPath(techProfiletblID uint32, uniPortName string) string {
+ return fmt.Sprintf(t.config.TPInstanceKVPath, t.resourceMgr.GetTechnology(), techProfiletblID, uniPortName)
+}
+
+func (t *TechProfileMgr) GetTPInstanceFromKVStore(ctx context.Context, techProfiletblID uint32, path string) (*TechProfile, error) {
+ var KvTpIns TechProfile
+ var resPtr *TechProfile = &KvTpIns
+ var err error
+ var kvResult *kvstore.KVPair
+
+ kvResult, _ = t.config.KVBackend.Get(ctx, path)
+ if kvResult == nil {
+ log.Infow("tp-instance-not-found-on-kv", log.Fields{"key": path})
+ return nil, nil
+ } else {
+ if value, err := kvstore.ToByte(kvResult.Value); err == nil {
+ if err = json.Unmarshal(value, resPtr); err != nil {
+ log.Errorw("error-unmarshal-kv-result", log.Fields{"key": path, "value": value})
+ return nil, errors.New("error-unmarshal-kv-result")
+ } else {
+ return resPtr, nil
+ }
+ }
+ }
+ return nil, err
+}
+
+func (t *TechProfileMgr) addTechProfInstanceToKVStore(ctx context.Context, techProfiletblID uint32, uniPortName string, tpInstance *TechProfile) error {
+ path := t.GetTechProfileInstanceKVPath(techProfiletblID, uniPortName)
+ log.Debugw("Adding techprof instance to kvstore", log.Fields{"key": path, "tpinstance": tpInstance})
+ tpInstanceJson, err := json.Marshal(*tpInstance)
+ if err == nil {
+ // Backend will convert JSON byte array into string format
+ log.Debugw("Storing tech profile instance to KV Store", log.Fields{"key": path, "val": tpInstanceJson})
+ err = t.config.KVBackend.Put(ctx, path, tpInstanceJson)
+ } else {
+ log.Errorw("Error in marshaling into Json format", log.Fields{"key": path, "tpinstance": tpInstance})
+ }
+ return err
+}
+func (t *TechProfileMgr) getTPFromKVStore(ctx context.Context, techProfiletblID uint32) *DefaultTechProfile {
+ var kvtechprofile DefaultTechProfile
+ key := fmt.Sprintf(t.config.TPFileKVPath, t.resourceMgr.GetTechnology(), techProfiletblID)
+ log.Debugw("Getting techprofile from KV store", log.Fields{"techProfiletblID": techProfiletblID, "Key": key})
+ kvresult, err := t.config.KVBackend.Get(ctx, key)
+ if err != nil {
+ log.Errorw("Error while fetching value from KV store", log.Fields{"key": key})
+ return nil
+ }
+ if kvresult != nil {
+ /* Backend will return Value in string format,needs to be converted to []byte before unmarshal*/
+ if value, err := kvstore.ToByte(kvresult.Value); err == nil {
+ if err = json.Unmarshal(value, &kvtechprofile); err != nil {
+ log.Errorw("Error unmarshaling techprofile fetched from KV store", log.Fields{"techProfiletblID": techProfiletblID, "error": err, "techprofile_json": value})
+ return nil
+ }
+
+ log.Debugw("Success fetched techprofile from KV store", log.Fields{"techProfiletblID": techProfiletblID, "value": kvtechprofile})
+ return &kvtechprofile
+ }
+ }
+ return nil
+}
+
+func (t *TechProfileMgr) CreateTechProfInstance(ctx context.Context, techProfiletblID uint32, uniPortName string, intfId uint32) (*TechProfile, error) {
+ var tpInstance *TechProfile
+ log.Infow("creating-tp-instance", log.Fields{"tableid": techProfiletblID, "uni": uniPortName, "intId": intfId})
+
+ // Make sure the uniPortName is as per format pon-{[0-9]+}/onu-{[0-9]+}/uni-{[0-9]+}
+ if !uniPortNameFormat.Match([]byte(uniPortName)) {
+ log.Errorw("uni-port-name-not-confirming-to-format", log.Fields{"uniPortName": uniPortName})
+ return nil, errors.New("uni-port-name-not-confirming-to-format")
+ }
+
+ tp := t.getTPFromKVStore(ctx, techProfiletblID)
+ if tp != nil {
+ if err := t.validateInstanceControlAttr(tp.InstanceCtrl); err != nil {
+ log.Error("invalid-instance-ctrl-attr--using-default-tp")
+ tp = t.getDefaultTechProfile()
+ } else {
+ log.Infow("using-specified-tp-from-kv-store", log.Fields{"tpid": techProfiletblID})
+ }
+ } else {
+ log.Info("tp-not-found-on-kv--creating-default-tp")
+ tp = t.getDefaultTechProfile()
+ }
+ tpInstancePath := t.GetTechProfileInstanceKVPath(techProfiletblID, uniPortName)
+ if tpInstance = t.allocateTPInstance(ctx, uniPortName, tp, intfId, tpInstancePath); tpInstance == nil {
+ log.Error("tp-intance-allocation-failed")
+ return nil, errors.New("tp-intance-allocation-failed")
+ }
+ if err := t.addTechProfInstanceToKVStore(ctx, techProfiletblID, uniPortName, tpInstance); err != nil {
+ log.Errorw("error-adding-tp-to-kv-store", log.Fields{"tableid": techProfiletblID, "uni": uniPortName})
+ return nil, errors.New("error-adding-tp-to-kv-store")
+ }
+ log.Infow("tp-added-to-kv-store-successfully",
+ log.Fields{"tpid": techProfiletblID, "uni": uniPortName, "intfId": intfId})
+ return tpInstance, nil
+}
+
+func (t *TechProfileMgr) DeleteTechProfileInstance(ctx context.Context, techProfiletblID uint32, uniPortName string) error {
+ path := t.GetTechProfileInstanceKVPath(techProfiletblID, uniPortName)
+ return t.config.KVBackend.Delete(ctx, path)
+}
+
+func (t *TechProfileMgr) validateInstanceControlAttr(instCtl InstanceControl) error {
+ if instCtl.Onu != "single-instance" && instCtl.Onu != "multi-instance" {
+ log.Errorw("invalid-onu-instance-control-attribute", log.Fields{"onu-inst": instCtl.Onu})
+ return errors.New("invalid-onu-instance-ctl-attr")
+ }
+
+ if instCtl.Uni != "single-instance" && instCtl.Uni != "multi-instance" {
+ log.Errorw("invalid-uni-instance-control-attribute", log.Fields{"uni-inst": instCtl.Uni})
+ return errors.New("invalid-uni-instance-ctl-attr")
+ }
+
+ if instCtl.Uni == "multi-instance" {
+ log.Error("uni-multi-instance-tp-not-supported")
+ return errors.New("uni-multi-instance-tp-not-supported")
+ }
+
+ return nil
+}
+
+func (t *TechProfileMgr) allocateTPInstance(ctx context.Context, uniPortName string, tp *DefaultTechProfile, intfId uint32, tpInstPath string) *TechProfile {
+
+ var usGemPortAttributeList []iGemPortAttribute
+ var dsGemPortAttributeList []iGemPortAttribute
+ var dsMulticastGemAttributeList []iGemPortAttribute
+ var dsUnicastGemAttributeList []iGemPortAttribute
+ var tcontIDs []uint32
+ var gemPorts []uint32
+ var err error
+
+ log.Infow("Allocating TechProfileMgr instance from techprofile template", log.Fields{"uniPortName": uniPortName, "intfId": intfId, "numGem": tp.NumGemPorts})
+
+ if tp.InstanceCtrl.Onu == "multi-instance" {
+ if tcontIDs, err = t.resourceMgr.GetResourceID(ctx, intfId, t.resourceMgr.GetResourceTypeAllocID(), 1); err != nil {
+ log.Errorw("Error getting alloc id from rsrcrMgr", log.Fields{"intfId": intfId})
+ return nil
+ }
+ } else { // "single-instance"
+ if tpInst, err := t.getSingleInstanceTp(ctx, tpInstPath); err != nil {
+ log.Errorw("Error getting alloc id from rsrcrMgr", log.Fields{"intfId": intfId})
+ return nil
+ } else if tpInst == nil {
+ // No "single-instance" tp found on one any uni port for the given TP ID
+ // Allocate a new TcontID or AllocID
+ if tcontIDs, err = t.resourceMgr.GetResourceID(ctx, intfId, t.resourceMgr.GetResourceTypeAllocID(), 1); err != nil {
+ log.Errorw("Error getting alloc id from rsrcrMgr", log.Fields{"intfId": intfId})
+ return nil
+ }
+ } else {
+ // Use the alloc-id from the existing TpInstance
+ tcontIDs = append(tcontIDs, tpInst.UsScheduler.AllocID)
+ }
+ }
+ log.Debugw("Num GEM ports in TP:", log.Fields{"NumGemPorts": tp.NumGemPorts})
+ if gemPorts, err = t.resourceMgr.GetResourceID(ctx, intfId, t.resourceMgr.GetResourceTypeGemPortID(), tp.NumGemPorts); err != nil {
+ log.Errorw("Error getting gemport ids from rsrcrMgr", log.Fields{"intfId": intfId, "numGemports": tp.NumGemPorts})
+ return nil
+ }
+ log.Infow("Allocated tconts and GEM ports successfully", log.Fields{"tconts": tcontIDs, "gemports": gemPorts})
+ for index := 0; index < int(tp.NumGemPorts); index++ {
+ usGemPortAttributeList = append(usGemPortAttributeList,
+ iGemPortAttribute{GemportID: gemPorts[index],
+ MaxQueueSize: tp.UpstreamGemPortAttributeList[index].MaxQueueSize,
+ PbitMap: tp.UpstreamGemPortAttributeList[index].PbitMap,
+ AesEncryption: tp.UpstreamGemPortAttributeList[index].AesEncryption,
+ SchedulingPolicy: tp.UpstreamGemPortAttributeList[index].SchedulingPolicy,
+ PriorityQueue: tp.UpstreamGemPortAttributeList[index].PriorityQueue,
+ Weight: tp.UpstreamGemPortAttributeList[index].Weight,
+ DiscardPolicy: tp.UpstreamGemPortAttributeList[index].DiscardPolicy,
+ DiscardConfig: tp.UpstreamGemPortAttributeList[index].DiscardConfig})
+ }
+
+ log.Info("length of DownstreamGemPortAttributeList", len(tp.DownstreamGemPortAttributeList))
+ //put multicast and unicast downstream GEM port attributes in different lists first
+ for index := 0; index < int(len(tp.DownstreamGemPortAttributeList)); index++ {
+ if isMulticastGem(tp.DownstreamGemPortAttributeList[index].IsMulticast) {
+ dsMulticastGemAttributeList = append(dsMulticastGemAttributeList,
+ iGemPortAttribute{
+ McastGemID: tp.DownstreamGemPortAttributeList[index].McastGemID,
+ MaxQueueSize: tp.DownstreamGemPortAttributeList[index].MaxQueueSize,
+ PbitMap: tp.DownstreamGemPortAttributeList[index].PbitMap,
+ AesEncryption: tp.DownstreamGemPortAttributeList[index].AesEncryption,
+ SchedulingPolicy: tp.DownstreamGemPortAttributeList[index].SchedulingPolicy,
+ PriorityQueue: tp.DownstreamGemPortAttributeList[index].PriorityQueue,
+ Weight: tp.DownstreamGemPortAttributeList[index].Weight,
+ DiscardPolicy: tp.DownstreamGemPortAttributeList[index].DiscardPolicy,
+ DiscardConfig: tp.DownstreamGemPortAttributeList[index].DiscardConfig,
+ IsMulticast: tp.DownstreamGemPortAttributeList[index].IsMulticast,
+ DControlList: tp.DownstreamGemPortAttributeList[index].DControlList,
+ SControlList: tp.DownstreamGemPortAttributeList[index].SControlList})
+ } else {
+ dsUnicastGemAttributeList = append(dsUnicastGemAttributeList,
+ iGemPortAttribute{
+ MaxQueueSize: tp.DownstreamGemPortAttributeList[index].MaxQueueSize,
+ PbitMap: tp.DownstreamGemPortAttributeList[index].PbitMap,
+ AesEncryption: tp.DownstreamGemPortAttributeList[index].AesEncryption,
+ SchedulingPolicy: tp.DownstreamGemPortAttributeList[index].SchedulingPolicy,
+ PriorityQueue: tp.DownstreamGemPortAttributeList[index].PriorityQueue,
+ Weight: tp.DownstreamGemPortAttributeList[index].Weight,
+ DiscardPolicy: tp.DownstreamGemPortAttributeList[index].DiscardPolicy,
+ DiscardConfig: tp.DownstreamGemPortAttributeList[index].DiscardConfig})
+ }
+ }
+ //add unicast downstream GEM ports to dsGemPortAttributeList
+ for index := 0; index < int(tp.NumGemPorts); index++ {
+ dsGemPortAttributeList = append(dsGemPortAttributeList,
+ iGemPortAttribute{GemportID: gemPorts[index],
+ MaxQueueSize: dsUnicastGemAttributeList[index].MaxQueueSize,
+ PbitMap: dsUnicastGemAttributeList[index].PbitMap,
+ AesEncryption: dsUnicastGemAttributeList[index].AesEncryption,
+ SchedulingPolicy: dsUnicastGemAttributeList[index].SchedulingPolicy,
+ PriorityQueue: dsUnicastGemAttributeList[index].PriorityQueue,
+ Weight: dsUnicastGemAttributeList[index].Weight,
+ DiscardPolicy: dsUnicastGemAttributeList[index].DiscardPolicy,
+ DiscardConfig: dsUnicastGemAttributeList[index].DiscardConfig})
+ }
+ //add multicast GEM ports to dsGemPortAttributeList afterwards
+ for k := range dsMulticastGemAttributeList {
+ dsGemPortAttributeList = append(dsGemPortAttributeList, dsMulticastGemAttributeList[k])
+ }
+
+ return &TechProfile{
+ SubscriberIdentifier: uniPortName,
+ Name: tp.Name,
+ ProfileType: tp.ProfileType,
+ Version: tp.Version,
+ NumGemPorts: tp.NumGemPorts,
+ InstanceCtrl: tp.InstanceCtrl,
+ UsScheduler: iScheduler{
+ AllocID: tcontIDs[0],
+ Direction: tp.UsScheduler.Direction,
+ AdditionalBw: tp.UsScheduler.AdditionalBw,
+ Priority: tp.UsScheduler.Priority,
+ Weight: tp.UsScheduler.Weight,
+ QSchedPolicy: tp.UsScheduler.QSchedPolicy},
+ DsScheduler: iScheduler{
+ AllocID: tcontIDs[0],
+ Direction: tp.DsScheduler.Direction,
+ AdditionalBw: tp.DsScheduler.AdditionalBw,
+ Priority: tp.DsScheduler.Priority,
+ Weight: tp.DsScheduler.Weight,
+ QSchedPolicy: tp.DsScheduler.QSchedPolicy},
+ UpstreamGemPortAttributeList: usGemPortAttributeList,
+ DownstreamGemPortAttributeList: dsGemPortAttributeList}
+}
+
+// getSingleInstanceTp returns another TpInstance for an ONU on a different
+// uni port for the same TP ID, if it finds one, else nil.
+func (t *TechProfileMgr) getSingleInstanceTp(ctx context.Context, tpPath string) (*TechProfile, error) {
+ var tpInst TechProfile
+
+ // For example:
+ // tpPath like "service/voltha/technology_profiles/xgspon/64/pon-{0}/onu-{1}/uni-{1}"
+ // is broken into ["service/voltha/technology_profiles/xgspon/64/pon-{0}/onu-{1}" ""]
+ uniPathSlice := regexp.MustCompile(`/uni-{[0-9]+}$`).Split(tpPath, 2)
+ kvPairs, _ := t.config.KVBackend.List(ctx, uniPathSlice[0])
+
+ // Find a valid TP Instance among all the UNIs of that ONU for the given TP ID
+ for keyPath, kvPair := range kvPairs {
+ if value, err := kvstore.ToByte(kvPair.Value); err == nil {
+ if err = json.Unmarshal(value, &tpInst); err != nil {
+ log.Errorw("error-unmarshal-kv-pair", log.Fields{"keyPath": keyPath, "value": value})
+ return nil, errors.New("error-unmarshal-kv-pair")
+ } else {
+ log.Debugw("found-valid-tp-instance-on-another-uni", log.Fields{"keyPath": keyPath})
+ return &tpInst, nil
+ }
+ }
+ }
+ return nil, nil
+}
+
+func (t *TechProfileMgr) getDefaultTechProfile() *DefaultTechProfile {
+
+ var usGemPortAttributeList []GemPortAttribute
+ var dsGemPortAttributeList []GemPortAttribute
+
+ for _, pbit := range t.config.DefaultPbits {
+ log.Debugw("Creating GEM port", log.Fields{"pbit": pbit})
+ usGemPortAttributeList = append(usGemPortAttributeList,
+ GemPortAttribute{
+ MaxQueueSize: defaultMaxQueueSize,
+ PbitMap: pbit,
+ AesEncryption: defaultAESEncryption,
+ SchedulingPolicy: SchedulingPolicy_name[defaultSchedulePolicy],
+ PriorityQueue: defaultPriorityQueue,
+ Weight: defaultQueueWeight,
+ DiscardPolicy: DiscardPolicy_name[defaultdropPolicy],
+ DiscardConfig: DiscardConfig{
+ MinThreshold: defaultMinThreshold,
+ MaxThreshold: defaultMaxThreshold,
+ MaxProbability: defaultMaxProbability}})
+ dsGemPortAttributeList = append(dsGemPortAttributeList,
+ GemPortAttribute{
+ MaxQueueSize: defaultMaxQueueSize,
+ PbitMap: pbit,
+ AesEncryption: defaultAESEncryption,
+ SchedulingPolicy: SchedulingPolicy_name[defaultSchedulePolicy],
+ PriorityQueue: defaultPriorityQueue,
+ Weight: defaultQueueWeight,
+ DiscardPolicy: DiscardPolicy_name[defaultdropPolicy],
+ DiscardConfig: DiscardConfig{
+ MinThreshold: defaultMinThreshold,
+ MaxThreshold: defaultMaxThreshold,
+ MaxProbability: defaultMaxProbability},
+ IsMulticast: defaultIsMulticast,
+ DControlList: defaultAccessControlList,
+ SControlList: defaultAccessControlList,
+ McastGemID: defaultMcastGemID})
+ }
+ return &DefaultTechProfile{
+ Name: t.config.DefaultTPName,
+ ProfileType: t.resourceMgr.GetTechnology(),
+ Version: t.config.TPVersion,
+ NumGemPorts: uint32(len(usGemPortAttributeList)),
+ InstanceCtrl: InstanceControl{
+ Onu: defaultOnuInstance,
+ Uni: defaultUniInstance,
+ MaxGemPayloadSize: defaultGemPayloadSize},
+ UsScheduler: Scheduler{
+ Direction: Direction_name[Direction_UPSTREAM],
+ AdditionalBw: AdditionalBW_name[defaultAdditionalBw],
+ Priority: defaultPriority,
+ Weight: defaultWeight,
+ QSchedPolicy: SchedulingPolicy_name[defaultQueueSchedPolicy]},
+ DsScheduler: Scheduler{
+ Direction: Direction_name[Direction_DOWNSTREAM],
+ AdditionalBw: AdditionalBW_name[defaultAdditionalBw],
+ Priority: defaultPriority,
+ Weight: defaultWeight,
+ QSchedPolicy: SchedulingPolicy_name[defaultQueueSchedPolicy]},
+ UpstreamGemPortAttributeList: usGemPortAttributeList,
+ DownstreamGemPortAttributeList: dsGemPortAttributeList}
+}
+
+func (t *TechProfileMgr) GetprotoBufParamValue(paramType string, paramKey string) int32 {
+ var result int32 = -1
+
+ if paramType == "direction" {
+ for key, val := range tp_pb.Direction_value {
+ if key == paramKey {
+ result = val
+ }
+ }
+ } else if paramType == "discard_policy" {
+ for key, val := range tp_pb.DiscardPolicy_value {
+ if key == paramKey {
+ result = val
+ }
+ }
+ } else if paramType == "sched_policy" {
+ for key, val := range tp_pb.SchedulingPolicy_value {
+ if key == paramKey {
+ log.Debugw("Got value in proto", log.Fields{"key": key, "value": val})
+ result = val
+ }
+ }
+ } else if paramType == "additional_bw" {
+ for key, val := range tp_pb.AdditionalBW_value {
+ if key == paramKey {
+ result = val
+ }
+ }
+ } else {
+ log.Error("Could not find proto parameter", log.Fields{"paramType": paramType, "key": paramKey})
+ return -1
+ }
+ log.Debugw("Got value in proto", log.Fields{"key": paramKey, "value": result})
+ return result
+}
+
+func (t *TechProfileMgr) GetUsScheduler(tpInstance *TechProfile) (*tp_pb.SchedulerConfig, error) {
+ dir := tp_pb.Direction(t.GetprotoBufParamValue("direction", tpInstance.UsScheduler.Direction))
+ if dir == -1 {
+ log.Errorf("Error in getting proto id for direction %s for upstream scheduler", tpInstance.UsScheduler.Direction)
+ return nil, fmt.Errorf("unable to get proto id for direction %s for upstream scheduler", tpInstance.UsScheduler.Direction)
+ }
+
+ bw := tp_pb.AdditionalBW(t.GetprotoBufParamValue("additional_bw", tpInstance.UsScheduler.AdditionalBw))
+ if bw == -1 {
+ log.Errorf("Error in getting proto id for bandwidth %s for upstream scheduler", tpInstance.UsScheduler.AdditionalBw)
+ return nil, fmt.Errorf("unable to get proto id for bandwidth %s for upstream scheduler", tpInstance.UsScheduler.AdditionalBw)
+ }
+
+ policy := tp_pb.SchedulingPolicy(t.GetprotoBufParamValue("sched_policy", tpInstance.UsScheduler.QSchedPolicy))
+ if policy == -1 {
+ log.Errorf("Error in getting proto id for scheduling policy %s for upstream scheduler", tpInstance.UsScheduler.QSchedPolicy)
+ return nil, fmt.Errorf("unable to get proto id for scheduling policy %s for upstream scheduler", tpInstance.UsScheduler.QSchedPolicy)
+ }
+
+ return &tp_pb.SchedulerConfig{
+ Direction: dir,
+ AdditionalBw: bw,
+ Priority: tpInstance.UsScheduler.Priority,
+ Weight: tpInstance.UsScheduler.Weight,
+ SchedPolicy: policy}, nil
+}
+
+func (t *TechProfileMgr) GetDsScheduler(tpInstance *TechProfile) (*tp_pb.SchedulerConfig, error) {
+
+ dir := tp_pb.Direction(t.GetprotoBufParamValue("direction", tpInstance.DsScheduler.Direction))
+ if dir == -1 {
+ log.Errorf("Error in getting proto id for direction %s for downstream scheduler", tpInstance.DsScheduler.Direction)
+ return nil, fmt.Errorf("unable to get proto id for direction %s for downstream scheduler", tpInstance.DsScheduler.Direction)
+ }
+
+ bw := tp_pb.AdditionalBW(t.GetprotoBufParamValue("additional_bw", tpInstance.DsScheduler.AdditionalBw))
+ if bw == -1 {
+ log.Errorf("Error in getting proto id for bandwidth %s for downstream scheduler", tpInstance.DsScheduler.AdditionalBw)
+ return nil, fmt.Errorf("unable to get proto id for bandwidth %s for downstream scheduler", tpInstance.DsScheduler.AdditionalBw)
+ }
+
+ policy := tp_pb.SchedulingPolicy(t.GetprotoBufParamValue("sched_policy", tpInstance.DsScheduler.QSchedPolicy))
+ if policy == -1 {
+ log.Errorf("Error in getting proto id for scheduling policy %s for downstream scheduler", tpInstance.DsScheduler.QSchedPolicy)
+ return nil, fmt.Errorf("unable to get proto id for scheduling policy %s for downstream scheduler", tpInstance.DsScheduler.QSchedPolicy)
+ }
+
+ return &tp_pb.SchedulerConfig{
+ Direction: dir,
+ AdditionalBw: bw,
+ Priority: tpInstance.DsScheduler.Priority,
+ Weight: tpInstance.DsScheduler.Weight,
+ SchedPolicy: policy}, nil
+}
+
+func (t *TechProfileMgr) GetTrafficScheduler(tpInstance *TechProfile, SchedCfg *tp_pb.SchedulerConfig,
+ ShapingCfg *tp_pb.TrafficShapingInfo) *tp_pb.TrafficScheduler {
+
+ tSched := &tp_pb.TrafficScheduler{
+ Direction: SchedCfg.Direction,
+ AllocId: tpInstance.UsScheduler.AllocID,
+ TrafficShapingInfo: ShapingCfg,
+ Scheduler: SchedCfg}
+
+ return tSched
+}
+
+func (tpm *TechProfileMgr) GetTrafficQueues(tp *TechProfile, Dir tp_pb.Direction) ([]*tp_pb.TrafficQueue, error) {
+
+ var encryp bool
+ if Dir == tp_pb.Direction_UPSTREAM {
+ // upstream GEM ports
+ NumGemPorts := len(tp.UpstreamGemPortAttributeList)
+ GemPorts := make([]*tp_pb.TrafficQueue, 0)
+ for Count := 0; Count < NumGemPorts; Count++ {
+ if tp.UpstreamGemPortAttributeList[Count].AesEncryption == "True" {
+ encryp = true
+ } else {
+ encryp = false
+ }
+
+ schedPolicy := tpm.GetprotoBufParamValue("sched_policy", tp.UpstreamGemPortAttributeList[Count].SchedulingPolicy)
+ if schedPolicy == -1 {
+ log.Errorf("Error in getting Proto Id for scheduling policy %s for Upstream Gem Port %d", tp.UpstreamGemPortAttributeList[Count].SchedulingPolicy, Count)
+ return nil, fmt.Errorf("upstream gem port traffic queue creation failed due to unrecognized scheduling policy %s", tp.UpstreamGemPortAttributeList[Count].SchedulingPolicy)
+ }
+
+ discardPolicy := tpm.GetprotoBufParamValue("discard_policy", tp.UpstreamGemPortAttributeList[Count].DiscardPolicy)
+ if discardPolicy == -1 {
+ log.Errorf("Error in getting Proto Id for discard policy %s for Upstream Gem Port %d", tp.UpstreamGemPortAttributeList[Count].DiscardPolicy, Count)
+ return nil, fmt.Errorf("upstream gem port traffic queue creation failed due to unrecognized discard policy %s", tp.UpstreamGemPortAttributeList[Count].DiscardPolicy)
+ }
+
+ GemPorts = append(GemPorts, &tp_pb.TrafficQueue{
+ Direction: tp_pb.Direction(tpm.GetprotoBufParamValue("direction", tp.UsScheduler.Direction)),
+ GemportId: tp.UpstreamGemPortAttributeList[Count].GemportID,
+ PbitMap: tp.UpstreamGemPortAttributeList[Count].PbitMap,
+ AesEncryption: encryp,
+ SchedPolicy: tp_pb.SchedulingPolicy(schedPolicy),
+ Priority: tp.UpstreamGemPortAttributeList[Count].PriorityQueue,
+ Weight: tp.UpstreamGemPortAttributeList[Count].Weight,
+ DiscardPolicy: tp_pb.DiscardPolicy(discardPolicy),
+ })
+ }
+ log.Debugw("Upstream Traffic queue list ", log.Fields{"queuelist": GemPorts})
+ return GemPorts, nil
+ } else if Dir == tp_pb.Direction_DOWNSTREAM {
+ //downstream GEM ports
+ NumGemPorts := len(tp.DownstreamGemPortAttributeList)
+ GemPorts := make([]*tp_pb.TrafficQueue, 0)
+ for Count := 0; Count < NumGemPorts; Count++ {
+ if isMulticastGem(tp.DownstreamGemPortAttributeList[Count].IsMulticast) {
+ //do not take multicast GEM ports. They are handled separately.
+ continue
+ }
+ if tp.DownstreamGemPortAttributeList[Count].AesEncryption == "True" {
+ encryp = true
+ } else {
+ encryp = false
+ }
+
+ schedPolicy := tpm.GetprotoBufParamValue("sched_policy", tp.DownstreamGemPortAttributeList[Count].SchedulingPolicy)
+ if schedPolicy == -1 {
+ log.Errorf("Error in getting Proto Id for scheduling policy %s for Downstream Gem Port %d", tp.DownstreamGemPortAttributeList[Count].SchedulingPolicy, Count)
+ return nil, fmt.Errorf("downstream gem port traffic queue creation failed due to unrecognized scheduling policy %s", tp.DownstreamGemPortAttributeList[Count].SchedulingPolicy)
+ }
+
+ discardPolicy := tpm.GetprotoBufParamValue("discard_policy", tp.DownstreamGemPortAttributeList[Count].DiscardPolicy)
+ if discardPolicy == -1 {
+ log.Errorf("Error in getting Proto Id for discard policy %s for Downstream Gem Port %d", tp.DownstreamGemPortAttributeList[Count].DiscardPolicy, Count)
+ return nil, fmt.Errorf("downstream gem port traffic queue creation failed due to unrecognized discard policy %s", tp.DownstreamGemPortAttributeList[Count].DiscardPolicy)
+ }
+
+ GemPorts = append(GemPorts, &tp_pb.TrafficQueue{
+ Direction: tp_pb.Direction(tpm.GetprotoBufParamValue("direction", tp.DsScheduler.Direction)),
+ GemportId: tp.DownstreamGemPortAttributeList[Count].GemportID,
+ PbitMap: tp.DownstreamGemPortAttributeList[Count].PbitMap,
+ AesEncryption: encryp,
+ SchedPolicy: tp_pb.SchedulingPolicy(schedPolicy),
+ Priority: tp.DownstreamGemPortAttributeList[Count].PriorityQueue,
+ Weight: tp.DownstreamGemPortAttributeList[Count].Weight,
+ DiscardPolicy: tp_pb.DiscardPolicy(discardPolicy),
+ })
+ }
+ log.Debugw("Downstream Traffic queue list ", log.Fields{"queuelist": GemPorts})
+ return GemPorts, nil
+ }
+
+ log.Errorf("Unsupported direction %s used for generating Traffic Queue list", Dir)
+ return nil, fmt.Errorf("downstream gem port traffic queue creation failed due to unsupported direction %s", Dir)
+}
+
+//isMulticastGem returns true if isMulticast attribute value of a GEM port is true; false otherwise
+func isMulticastGem(isMulticastAttrValue string) bool {
+ return isMulticastAttrValue != "" &&
+ (isMulticastAttrValue == "True" || isMulticastAttrValue == "true" || isMulticastAttrValue == "TRUE")
+}
+
+func (tpm *TechProfileMgr) GetMulticastTrafficQueues(tp *TechProfile) []*tp_pb.TrafficQueue {
+ var encryp bool
+ NumGemPorts := len(tp.DownstreamGemPortAttributeList)
+ mcastTrafficQueues := make([]*tp_pb.TrafficQueue, 0)
+ for Count := 0; Count < NumGemPorts; Count++ {
+ if !isMulticastGem(tp.DownstreamGemPortAttributeList[Count].IsMulticast) {
+ continue
+ }
+ if tp.DownstreamGemPortAttributeList[Count].AesEncryption == "True" {
+ encryp = true
+ } else {
+ encryp = false
+ }
+ mcastTrafficQueues = append(mcastTrafficQueues, &tp_pb.TrafficQueue{
+ Direction: tp_pb.Direction(tpm.GetprotoBufParamValue("direction", tp.DsScheduler.Direction)),
+ GemportId: tp.DownstreamGemPortAttributeList[Count].McastGemID,
+ PbitMap: tp.DownstreamGemPortAttributeList[Count].PbitMap,
+ AesEncryption: encryp,
+ SchedPolicy: tp_pb.SchedulingPolicy(tpm.GetprotoBufParamValue("sched_policy", tp.DownstreamGemPortAttributeList[Count].SchedulingPolicy)),
+ Priority: tp.DownstreamGemPortAttributeList[Count].PriorityQueue,
+ Weight: tp.DownstreamGemPortAttributeList[Count].Weight,
+ DiscardPolicy: tp_pb.DiscardPolicy(tpm.GetprotoBufParamValue("discard_policy", tp.DownstreamGemPortAttributeList[Count].DiscardPolicy)),
+ })
+ }
+ log.Debugw("Downstream Multicast Traffic queue list ", log.Fields{"queuelist": mcastTrafficQueues})
+ return mcastTrafficQueues
+}
+
+func (tpm *TechProfileMgr) GetUsTrafficScheduler(tp *TechProfile) *tp_pb.TrafficScheduler {
+ UsScheduler, _ := tpm.GetUsScheduler(tp)
+
+ return &tp_pb.TrafficScheduler{Direction: UsScheduler.Direction,
+ AllocId: tp.UsScheduler.AllocID,
+ Scheduler: UsScheduler}
+}
+
+func (t *TechProfileMgr) GetGemportIDForPbit(tp *TechProfile, Dir tp_pb.Direction, pbit uint32) uint32 {
+ /*
+ Function to get the Gemport ID mapped to a pbit.
+ */
+ if Dir == tp_pb.Direction_UPSTREAM {
+ // upstream GEM ports
+ NumGemPorts := len(tp.UpstreamGemPortAttributeList)
+ for Count := 0; Count < NumGemPorts; Count++ {
+ NumPbitMaps := len(tp.UpstreamGemPortAttributeList[Count].PbitMap)
+ for ICount := 2; ICount < NumPbitMaps; ICount++ {
+ if p, err := strconv.Atoi(string(tp.UpstreamGemPortAttributeList[Count].PbitMap[ICount])); err == nil {
+ if uint32(ICount-2) == pbit && p == 1 { // Check this p-bit is set
+ log.Debugw("Found-US-GEMport-for-Pcp", log.Fields{"pbit": pbit, "GEMport": tp.UpstreamGemPortAttributeList[Count].GemportID})
+ return tp.UpstreamGemPortAttributeList[Count].GemportID
+ }
+ }
+ }
+ }
+ } else if Dir == tp_pb.Direction_DOWNSTREAM {
+ //downstream GEM ports
+ NumGemPorts := len(tp.DownstreamGemPortAttributeList)
+ for Count := 0; Count < NumGemPorts; Count++ {
+ NumPbitMaps := len(tp.DownstreamGemPortAttributeList[Count].PbitMap)
+ for ICount := 2; ICount < NumPbitMaps; ICount++ {
+ if p, err := strconv.Atoi(string(tp.DownstreamGemPortAttributeList[Count].PbitMap[ICount])); err == nil {
+ if uint32(ICount-2) == pbit && p == 1 { // Check this p-bit is set
+ log.Debugw("Found-DS-GEMport-for-Pcp", log.Fields{"pbit": pbit, "GEMport": tp.DownstreamGemPortAttributeList[Count].GemportID})
+ return tp.DownstreamGemPortAttributeList[Count].GemportID
+ }
+ }
+ }
+ }
+ }
+ log.Errorw("No-GemportId-Found-For-Pcp", log.Fields{"pcpVlan": pbit})
+ return 0
+}
+
+// FindAllTpInstances returns all TechProfile instances for a given TechProfile table-id, pon interface ID and onu ID.
+func (t *TechProfileMgr) FindAllTpInstances(ctx context.Context, techProfiletblID uint32, ponIntf uint32, onuID uint32) []TechProfile {
+ var tp TechProfile
+ onuTpInstancePath := fmt.Sprintf("%s/%d/pon-{%d}/onu-{%d}", t.resourceMgr.GetTechnology(), techProfiletblID, ponIntf, onuID)
+
+ if kvPairs, _ := t.config.KVBackend.List(ctx, onuTpInstancePath); kvPairs != nil {
+ tpInstances := make([]TechProfile, 0, len(kvPairs))
+ for kvPath, kvPair := range kvPairs {
+ if value, err := kvstore.ToByte(kvPair.Value); err == nil {
+ if err = json.Unmarshal(value, &tp); err != nil {
+ log.Errorw("error-unmarshal-kv-pair", log.Fields{"kvPath": kvPath, "value": value})
+ continue
+ } else {
+ tpInstances = append(tpInstances, tp)
+ }
+ }
+ }
+ return tpInstances
+ }
+ return nil
+}
diff --git a/vendor/github.com/opencord/voltha-lib-go/v3/pkg/techprofile/tech_profile_if.go b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/techprofile/tech_profile_if.go
new file mode 100644
index 0000000..e605d49
--- /dev/null
+++ b/vendor/github.com/opencord/voltha-lib-go/v3/pkg/techprofile/tech_profile_if.go
@@ -0,0 +1,41 @@
+/*
+ * Copyright 2019-present Open Networking Foundation
+
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+
+ * http://www.apache.org/licenses/LICENSE-2.0
+
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package techprofile
+
+import (
+ "context"
+
+ "github.com/opencord/voltha-lib-go/v3/pkg/db"
+ tp_pb "github.com/opencord/voltha-protos/v3/go/tech_profile"
+)
+
+type TechProfileIf interface {
+ SetKVClient() *db.Backend
+ GetTechProfileInstanceKVPath(techProfiletblID uint32, uniPortName string) string
+ GetTPInstanceFromKVStore(ctx context.Context, techProfiletblID uint32, path string) (*TechProfile, error)
+ CreateTechProfInstance(ctx context.Context, techProfiletblID uint32, uniPortName string, intfId uint32) (*TechProfile, error)
+ DeleteTechProfileInstance(ctx context.Context, techProfiletblID uint32, uniPortName string) error
+ GetprotoBufParamValue(paramType string, paramKey string) int32
+ GetUsScheduler(tpInstance *TechProfile) (*tp_pb.SchedulerConfig, error)
+ GetDsScheduler(tpInstance *TechProfile) (*tp_pb.SchedulerConfig, error)
+ GetTrafficScheduler(tpInstance *TechProfile, SchedCfg *tp_pb.SchedulerConfig,
+ ShapingCfg *tp_pb.TrafficShapingInfo) *tp_pb.TrafficScheduler
+ GetTrafficQueues(tp *TechProfile, Dir tp_pb.Direction) ([]*tp_pb.TrafficQueue, error)
+ GetMulticastTrafficQueues(tp *TechProfile) []*tp_pb.TrafficQueue
+ GetGemportIDForPbit(tp *TechProfile, Dir tp_pb.Direction, pbit uint32) uint32
+ FindAllTpInstances(ctx context.Context, techProfiletblID uint32, ponIntf uint32, onuID uint32) []TechProfile
+}