VOL-4217: Update to latest version of protos and voltha-lib-go for
voltha-2.9 release. Tested with BAL3.10
Change-Id: Ibdc5978a1f2df713965a27ce26d0e22c1ffa366a
diff --git a/vendor/github.com/opencord/voltha-lib-go/v4/pkg/db/kvstore/consulclient.go b/vendor/github.com/opencord/voltha-lib-go/v4/pkg/db/kvstore/consulclient.go
deleted file mode 100644
index 2593608..0000000
--- a/vendor/github.com/opencord/voltha-lib-go/v4/pkg/db/kvstore/consulclient.go
+++ /dev/null
@@ -1,512 +0,0 @@
-/*
- * 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/v4/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(ctx context.Context, addr string, timeout time.Duration) (*ConsulClient, error) {
- config := consulapi.DefaultConfig()
- config.Address = addr
- config.WaitTime = timeout
- consul, err := consulapi.NewClient(config)
- if err != nil {
- logger.Error(ctx, 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(ctx, "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
- // Substract current time from deadline to get the waitTime duration
- queryOptions.WaitTime = time.Until(deadline)
-
- // For now we ignore meta data
- kvps, _, err := kv.List(key, &queryOptions)
- if err != nil {
- logger.Error(ctx, 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
- // Substract current time from deadline to get the waitTime duration
- queryOptions.WaitTime = time.Until(deadline)
-
- // For now we ignore meta data
- kvp, _, err := kv.Get(key, &queryOptions)
- if err != nil {
- logger.Error(ctx, 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(ctx, 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(ctx, 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(ctx, err)
- return err
- }
- return nil
-}
-
-func (c *ConsulClient) deleteSession(ctx context.Context) {
- if c.sessionID != "" {
- logger.Debug(ctx, "cleaning-up-session")
- session := c.consul.Session()
- _, err := session.Destroy(c.sessionID, nil)
- if err != nil {
- logger.Errorw(ctx, "error-cleaning-session", log.Fields{"session": c.sessionID, "error": err})
- }
- }
- c.sessionID = ""
- c.session = nil
-}
-
-func (c *ConsulClient) createSession(ctx context.Context, ttl time.Duration, retries int) (*consulapi.Session, string, error) {
- session := c.consul.Session()
- entry := &consulapi.SessionEntry{
- Behavior: consulapi.SessionBehaviorDelete,
- TTL: ttl.String(),
- }
-
- for {
- id, meta, err := session.Create(entry, nil)
- if err != nil {
- logger.Errorw(ctx, "create-session-error", log.Fields{"error": err})
- if retries == 0 {
- return nil, "", err
- }
- } else if meta.RequestTime == 0 {
- logger.Errorw(ctx, "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(ctx, "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(ctx, "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 time.Duration) (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(ctx, er)
- return nil, er
- }
-
- // Cleanup any existing session and recreate new ones. A key is reserved against a session
- if c.sessionID != "" {
- c.deleteSession(ctx)
- }
-
- // Clear session if reservation is not successful
- reservationSuccessful := false
- defer func() {
- if !reservationSuccessful {
- logger.Debug(ctx, "deleting-session")
- c.deleteSession(ctx)
- }
- }()
-
- session, sessionID, err := c.createSession(ctx, ttl, -1)
- if err != nil {
- logger.Errorw(ctx, "no-session-created", log.Fields{"error": err})
- return "", errors.New("no-session-created")
- }
- logger.Debugw(ctx, "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(ctx, "error-acquiring-keys", log.Fields{"error": err})
- return nil, err
- }
-
- logger.Debugw(ctx, "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(ctx, "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(ctx, "cannot-release-reservation", log.Fields{"key": key, "error": err})
- return err
- }
- if !result {
- logger.Errorw(ctx, "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(ctx context.Context, 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(ctx, "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(ctx, "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(ctx, "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(ctx context.Context, key string, ch chan *Event) {
- logger.Debugw(ctx, "start-watching-channel", log.Fields{"key": key, "channel": ch})
-
- defer c.CloseWatch(ctx, key, ch)
- kv := c.consul.KV()
- var queryOptions consulapi.QueryOptions
- queryOptions.WaitTime = defaultKVGetTimeout
-
- // Get the existing value, if any
- previousKVPair, meta, err := kv.Get(key, &queryOptions)
- if err != nil {
- logger.Debug(ctx, 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(ctx)
- for {
- //waitOptions = consulapi.QueryOptions{WaitIndex: lastIndex}
- waitOptions.WaitIndex = lastIndex
- pair, meta, err = kv.Get(key, waitOptions)
- select {
- case <-ctx.Done():
- logger.Debug(ctx, "done-event-received-exiting")
- return
- default:
- if err != nil {
- logger.Warnw(ctx, "error-from-watch", log.Fields{"error": err})
- ch <- NewEvent(CONNECTIONDOWN, key, []byte(""), -1)
- } else {
- logger.Debugw(ctx, "index-state", log.Fields{"lastindex": lastIndex, "newindex": meta.LastIndex, "key": key})
- }
- }
- if err != nil {
- logger.Debug(ctx, err)
- // On error, block for 10 milliseconds to prevent endless loop
- time.Sleep(10 * time.Millisecond)
- } else if meta.LastIndex <= lastIndex {
- logger.Info(ctx, "no-index-change-or-negative")
- } else {
- logger.Debugw(ctx, "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(ctx, "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(ctx context.Context) {
- 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(ctx, "error-closing-client", log.Fields{"error": err})
- }
-}
-
-func (c *ConsulClient) AcquireLock(ctx context.Context, lockName string, timeout time.Duration) error {
- return nil
-}
-
-func (c *ConsulClient) ReleaseLock(lockName string) error {
- return nil
-}
diff --git a/vendor/github.com/opencord/voltha-lib-go/v4/pkg/db/kvstore/etcdclient.go b/vendor/github.com/opencord/voltha-lib-go/v4/pkg/db/kvstore/etcdclient.go
deleted file mode 100644
index aa5adbf..0000000
--- a/vendor/github.com/opencord/voltha-lib-go/v4/pkg/db/kvstore/etcdclient.go
+++ /dev/null
@@ -1,487 +0,0 @@
-/*
- * 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"
- "time"
-
- "github.com/opencord/voltha-lib-go/v4/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(ctx context.Context, addr string, timeout time.Duration, level log.LogLevel) (*EtcdClient, error) {
- logconfig := log.ConstructZapConfig(log.JSON, level, log.Fields{})
-
- c, err := v3Client.New(v3Client.Config{
- Endpoints: []string{addr},
- DialTimeout: timeout,
- LogConfig: &logconfig,
- })
- if err != nil {
- logger.Error(ctx, 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(ctx, 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(ctx, 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(ctx, "context-cancelled", log.Fields{"error": err})
- case context.DeadlineExceeded:
- logger.Warnw(ctx, "context-deadline-exceeded", log.Fields{"error": err})
- case v3rpcTypes.ErrEmptyKey:
- logger.Warnw(ctx, "etcd-client-error", log.Fields{"error": err})
- default:
- logger.Warnw(ctx, "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(ctx, "failed-to-delete-key", log.Fields{"key": key, "error": err})
- return err
- }
- logger.Debugw(ctx, "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 time.Duration) (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, int64(ttl.Seconds()))
- if err != nil {
- logger.Error(ctx, 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(ctx, "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(ctx, "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(ctx, "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(ctx, 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(ctx, "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(ctx, "watched-channels", log.Fields{"len": len(channelMaps)})
- // Launch a go routine to listen for updates
- go c.listenForKeyChange(ctx, 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(ctx context.Context, 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(ctx, "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(ctx, "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(ctx, "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(ctx, "watcher-channel-exiting", log.Fields{"key": key, "channel": channelMaps})
-}
-
-func (c *EtcdClient) listenForKeyChange(ctx context.Context, channel v3Client.WatchChan, ch chan<- *Event, cancel context.CancelFunc) {
- logger.Debug(ctx, "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(ctx, "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(ctx context.Context) {
- if err := c.ectdAPI.Close(); err != nil {
- logger.Errorw(ctx, "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 time.Duration) 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/v4/pkg/techprofile/tech_profile.go b/vendor/github.com/opencord/voltha-lib-go/v4/pkg/techprofile/tech_profile.go
deleted file mode 100644
index b80f136..0000000
--- a/vendor/github.com/opencord/voltha-lib-go/v4/pkg/techprofile/tech_profile.go
+++ /dev/null
@@ -1,1400 +0,0 @@
-/*
- * 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"
- "sync"
- "time"
-
- "github.com/opencord/voltha-lib-go/v4/pkg/db"
-
- "github.com/opencord/voltha-lib-go/v4/pkg/db/kvstore"
- "github.com/opencord/voltha-lib-go/v4/pkg/log"
- tp_pb "github.com/opencord/voltha-protos/v4/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(`^olt-{[a-z0-9\-]+}/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"`
-}
-
-// Instance of Scheduler
-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"`
-}
-
-// Instance of GemPortAttribute
-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
- GemPortIDMgmtLock sync.RWMutex
- AllocIDMgmtLock sync.RWMutex
-}
-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"`
-}
-
-// QThresholds struct for EPON
-type QThresholds struct {
- QThreshold1 uint32 `json:"q_threshold1"`
- QThreshold2 uint32 `json:"q_threshold2"`
- QThreshold3 uint32 `json:"q_threshold3"`
- QThreshold4 uint32 `json:"q_threshold4"`
- QThreshold5 uint32 `json:"q_threshold5"`
- QThreshold6 uint32 `json:"q_threshold6"`
- QThreshold7 uint32 `json:"q_threshold7"`
-}
-
-// UpstreamQueueAttribute struct for EPON
-type UpstreamQueueAttribute struct {
- MaxQueueSize string `json:"max_q_size"`
- PbitMap string `json:"pbit_map"`
- AesEncryption string `json:"aes_encryption"`
- TrafficType string `json:"traffic_type"`
- UnsolicitedGrantSize uint32 `json:"unsolicited_grant_size"`
- NominalInterval uint32 `json:"nominal_interval"`
- ToleratedPollJitter uint32 `json:"tolerated_poll_jitter"`
- RequestTransmissionPolicy uint32 `json:"request_transmission_policy"`
- NumQueueSet uint32 `json:"num_q_sets"`
- QThresholds QThresholds `json:"q_thresholds"`
- SchedulingPolicy string `json:"scheduling_policy"`
- PriorityQueue uint32 `json:"priority_q"`
- Weight uint32 `json:"weight"`
- DiscardPolicy string `json:"discard_policy"`
- DiscardConfig DiscardConfig `json:"discard_config"`
-}
-
-// Default EPON constants
-const (
- defaultPakageType = "B"
-)
-const (
- defaultTrafficType = "BE"
- defaultUnsolicitedGrantSize = 0
- defaultNominalInterval = 0
- defaultToleratedPollJitter = 0
- defaultRequestTransmissionPolicy = 0
- defaultNumQueueSet = 2
-)
-const (
- defaultQThreshold1 = 5500
- defaultQThreshold2 = 0
- defaultQThreshold3 = 0
- defaultQThreshold4 = 0
- defaultQThreshold5 = 0
- defaultQThreshold6 = 0
- defaultQThreshold7 = 0
-)
-
-// DownstreamQueueAttribute struct for EPON
-type DownstreamQueueAttribute 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"`
-}
-
-// iUpstreamQueueAttribute struct for EPON
-type iUpstreamQueueAttribute struct {
- GemportID uint32 `json:"q_id"`
- MaxQueueSize string `json:"max_q_size"`
- PbitMap string `json:"pbit_map"`
- AesEncryption string `json:"aes_encryption"`
- TrafficType string `json:"traffic_type"`
- UnsolicitedGrantSize uint32 `json:"unsolicited_grant_size"`
- NominalInterval uint32 `json:"nominal_interval"`
- ToleratedPollJitter uint32 `json:"tolerated_poll_jitter"`
- RequestTransmissionPolicy uint32 `json:"request_transmission_policy"`
- NumQueueSet uint32 `json:"num_q_sets"`
- QThresholds QThresholds `json:"q_thresholds"`
- SchedulingPolicy string `json:"scheduling_policy"`
- PriorityQueue uint32 `json:"priority_q"`
- Weight uint32 `json:"weight"`
- DiscardPolicy string `json:"discard_policy"`
- DiscardConfig DiscardConfig `json:"discard_config"`
-}
-
-// iDownstreamQueueAttribute struct for EPON
-type iDownstreamQueueAttribute struct {
- GemportID uint32 `json:"q_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"`
-}
-
-// EponAttribute struct for EPON
-type EponAttribute struct {
- PackageType string `json:"pakage_type"`
-}
-
-// DefaultTechProfile struct for EPON
-type DefaultEponProfile 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"`
- EponAttribute EponAttribute `json:"epon_attribute"`
- UpstreamQueueAttributeList []UpstreamQueueAttribute `json:"upstream_queue_attribute_list"`
- DownstreamQueueAttributeList []DownstreamQueueAttribute `json:"downstream_queue_attribute_list"`
-}
-
-// TechProfile struct for EPON
-type EponProfile 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"`
- EponAttribute EponAttribute `json:"epon_attribute"`
- AllocID uint32 `json:"llid"`
- UpstreamQueueAttributeList []iUpstreamQueueAttribute `json:"upstream_queue_attribute_list"`
- DownstreamQueueAttributeList []iDownstreamQueueAttribute `json:"downstream_queue_attribute_list"`
-}
-
-const (
- xgspon = "XGS-PON"
- gpon = "GPON"
- epon = "EPON"
-)
-
-func (t *TechProfileMgr) SetKVClient(ctx context.Context) *db.Backend {
- kvClient, err := newKVClient(ctx, t.config.KVStoreType, t.config.KVStoreAddress, t.config.KVStoreTimeout)
- if err != nil {
- logger.Errorw(ctx, "failed-to-create-kv-client",
- log.Fields{
- "type": t.config.KVStoreType, "address": t.config.KVStoreAddress,
- "timeout": t.config.KVStoreTimeout, "prefix": t.config.TPKVPathPrefix,
- "error": err.Error(),
- })
- return nil
- }
- return &db.Backend{
- Client: kvClient,
- StoreType: t.config.KVStoreType,
- Address: t.config.KVStoreAddress,
- 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(ctx context.Context, storeType string, address string, timeout time.Duration) (kvstore.Client, error) {
-
- logger.Infow(ctx, "kv-store", log.Fields{"storeType": storeType, "address": address})
- switch storeType {
- case "consul":
- return kvstore.NewConsulClient(ctx, address, timeout)
- case "etcd":
- return kvstore.NewEtcdClient(ctx, address, timeout, log.WarnLevel)
- }
- return nil, errors.New("unsupported-kv-store")
-}
-
-func NewTechProfile(ctx context.Context, resourceMgr iPonResourceMgr, KVStoreType string, KVStoreAddress string) (*TechProfileMgr, error) {
- var techprofileObj TechProfileMgr
- logger.Debug(ctx, "Initializing techprofile Manager")
- techprofileObj.config = NewTechProfileFlags(KVStoreType, KVStoreAddress)
- techprofileObj.config.KVBackend = techprofileObj.SetKVClient(ctx)
- if techprofileObj.config.KVBackend == nil {
- logger.Error(ctx, "Failed to initialize KV backend\n")
- return nil, errors.New("KV backend init failed")
- }
- techprofileObj.resourceMgr = resourceMgr
- logger.Debug(ctx, "Initializing techprofile object instance success")
- return &techprofileObj, nil
-}
-
-func (t *TechProfileMgr) GetTechProfileInstanceKVPath(ctx context.Context, techProfiletblID uint32, uniPortName string) string {
- logger.Debugw(ctx, "get-tp-instance-kv-path", log.Fields{
- "uniPortName": uniPortName,
- "tpId": techProfiletblID,
- })
- return fmt.Sprintf(t.config.TPInstanceKVPath, t.resourceMgr.GetTechnology(), techProfiletblID, uniPortName)
-}
-
-func (t *TechProfileMgr) GetTPInstanceFromKVStore(ctx context.Context, techProfiletblID uint32, path string) (interface{}, error) {
- var err error
- var kvResult *kvstore.KVPair
- var KvTpIns TechProfile
- var KvEponIns EponProfile
- var resPtr interface{}
- // For example:
- // tpInstPath like "XGS-PON/64/uni_port_name"
- // is broken into ["XGS-PON" "64" ...]
- pathSlice := regexp.MustCompile(`/`).Split(path, -1)
- switch pathSlice[0] {
- case xgspon, gpon:
- resPtr = &KvTpIns
- case epon:
- resPtr = &KvEponIns
- default:
- logger.Errorw(ctx, "unknown-tech", log.Fields{"tech": pathSlice[0]})
- return nil, fmt.Errorf("unknown-tech-%s", pathSlice[0])
- }
-
- kvResult, _ = t.config.KVBackend.Get(ctx, path)
- if kvResult == nil {
- logger.Infow(ctx, "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 {
- logger.Errorw(ctx, "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(ctx, techProfiletblID, uniPortName)
- logger.Debugw(ctx, "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
- logger.Debugw(ctx, "Storing tech profile instance to KV Store", log.Fields{"key": path, "val": tpInstanceJson})
- err = t.config.KVBackend.Put(ctx, path, tpInstanceJson)
- } else {
- logger.Errorw(ctx, "Error in marshaling into Json format", log.Fields{"key": path, "tpinstance": tpInstance})
- }
- return err
-}
-
-func (t *TechProfileMgr) addEponProfInstanceToKVStore(ctx context.Context, techProfiletblID uint32, uniPortName string, tpInstance *EponProfile) error {
- path := t.GetTechProfileInstanceKVPath(ctx, techProfiletblID, uniPortName)
- logger.Debugw(ctx, "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
- logger.Debugw(ctx, "Storing tech profile instance to KV Store", log.Fields{"key": path, "val": tpInstanceJson})
- err = t.config.KVBackend.Put(ctx, path, tpInstanceJson)
- } else {
- logger.Errorw(ctx, "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)
- logger.Debugw(ctx, "Getting techprofile from KV store", log.Fields{"techProfiletblID": techProfiletblID, "Key": key})
- kvresult, err := t.config.KVBackend.Get(ctx, key)
- if err != nil {
- logger.Errorw(ctx, "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 {
- logger.Errorw(ctx, "Error unmarshaling techprofile fetched from KV store", log.Fields{"techProfiletblID": techProfiletblID, "error": err, "techprofile_json": value})
- return nil
- }
-
- logger.Debugw(ctx, "Success fetched techprofile from KV store", log.Fields{"techProfiletblID": techProfiletblID, "value": kvtechprofile})
- return &kvtechprofile
- }
- }
- return nil
-}
-
-func (t *TechProfileMgr) getEponTPFromKVStore(ctx context.Context, techProfiletblID uint32) *DefaultEponProfile {
- var kvtechprofile DefaultEponProfile
- key := fmt.Sprintf(t.config.TPFileKVPath, t.resourceMgr.GetTechnology(), techProfiletblID)
- logger.Debugw(ctx, "Getting techprofile from KV store", log.Fields{"techProfiletblID": techProfiletblID, "Key": key})
- kvresult, err := t.config.KVBackend.Get(ctx, key)
- if err != nil {
- logger.Errorw(ctx, "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 {
- logger.Errorw(ctx, "Error unmarshaling techprofile fetched from KV store", log.Fields{"techProfiletblID": techProfiletblID, "error": err, "techprofile_json": value})
- return nil
- }
-
- logger.Debugw(ctx, "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) (interface{}, error) {
- var tpInstance *TechProfile
- var tpEponInstance *EponProfile
-
- logger.Infow(ctx, "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)) {
- logger.Errorw(ctx, "uni-port-name-not-confirming-to-format", log.Fields{"uniPortName": uniPortName})
- return nil, errors.New("uni-port-name-not-confirming-to-format")
- }
- tpInstancePath := t.GetTechProfileInstanceKVPath(ctx, techProfiletblID, uniPortName)
- // For example:
- // tpInstPath like "XGS-PON/64/uni_port_name"
- // is broken into ["XGS-PON" "64" ...]
- pathSlice := regexp.MustCompile(`/`).Split(tpInstancePath, -1)
- if pathSlice[0] == epon {
- tp := t.getEponTPFromKVStore(ctx, techProfiletblID)
- if tp != nil {
- if err := t.validateInstanceControlAttr(ctx, tp.InstanceCtrl); err != nil {
- logger.Error(ctx, "invalid-instance-ctrl-attr--using-default-tp")
- tp = t.getDefaultEponProfile(ctx)
- } else {
- logger.Infow(ctx, "using-specified-tp-from-kv-store", log.Fields{"tpid": techProfiletblID})
- }
- } else {
- logger.Info(ctx, "tp-not-found-on-kv--creating-default-tp")
- tp = t.getDefaultEponProfile(ctx)
- }
-
- if tpEponInstance = t.allocateEponTPInstance(ctx, uniPortName, tp, intfId, tpInstancePath); tpEponInstance == nil {
- logger.Error(ctx, "tp-intance-allocation-failed")
- return nil, errors.New("tp-intance-allocation-failed")
- }
- if err := t.addEponProfInstanceToKVStore(ctx, techProfiletblID, uniPortName, tpEponInstance); err != nil {
- logger.Errorw(ctx, "error-adding-tp-to-kv-store", log.Fields{"tableid": techProfiletblID, "uni": uniPortName})
- return nil, errors.New("error-adding-tp-to-kv-store")
- }
- logger.Infow(ctx, "tp-added-to-kv-store-successfully",
- log.Fields{"tpid": techProfiletblID, "uni": uniPortName, "intfId": intfId})
- return tpEponInstance, nil
- } else {
- tp := t.getTPFromKVStore(ctx, techProfiletblID)
- if tp != nil {
- if err := t.validateInstanceControlAttr(ctx, tp.InstanceCtrl); err != nil {
- logger.Error(ctx, "invalid-instance-ctrl-attr--using-default-tp")
- tp = t.getDefaultTechProfile(ctx)
- } else {
- logger.Infow(ctx, "using-specified-tp-from-kv-store", log.Fields{"tpid": techProfiletblID})
- }
- } else {
- logger.Info(ctx, "tp-not-found-on-kv--creating-default-tp")
- tp = t.getDefaultTechProfile(ctx)
- }
-
- if tpInstance = t.allocateTPInstance(ctx, uniPortName, tp, intfId, tpInstancePath); tpInstance == nil {
- logger.Error(ctx, "tp-intance-allocation-failed")
- return nil, errors.New("tp-intance-allocation-failed")
- }
- if err := t.addTechProfInstanceToKVStore(ctx, techProfiletblID, uniPortName, tpInstance); err != nil {
- logger.Errorw(ctx, "error-adding-tp-to-kv-store", log.Fields{"tableid": techProfiletblID, "uni": uniPortName})
- return nil, errors.New("error-adding-tp-to-kv-store")
- }
- logger.Infow(ctx, "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(ctx, techProfiletblID, uniPortName)
- return t.config.KVBackend.Delete(ctx, path)
-}
-
-func (t *TechProfileMgr) validateInstanceControlAttr(ctx context.Context, instCtl InstanceControl) error {
- if instCtl.Onu != "single-instance" && instCtl.Onu != "multi-instance" {
- logger.Errorw(ctx, "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" {
- logger.Errorw(ctx, "invalid-uni-instance-control-attribute", log.Fields{"uni-inst": instCtl.Uni})
- return errors.New("invalid-uni-instance-ctl-attr")
- }
-
- if instCtl.Uni == "multi-instance" {
- logger.Error(ctx, "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
-
- logger.Infow(ctx, "Allocating TechProfileMgr instance from techprofile template", log.Fields{"uniPortName": uniPortName, "intfId": intfId, "numGem": tp.NumGemPorts})
-
- if tp.InstanceCtrl.Onu == "multi-instance" {
- t.AllocIDMgmtLock.Lock()
- tcontIDs, err = t.resourceMgr.GetResourceID(ctx, intfId, t.resourceMgr.GetResourceTypeAllocID(), 1)
- t.AllocIDMgmtLock.Unlock()
- if err != nil {
- logger.Errorw(ctx, "Error getting alloc id from rsrcrMgr", log.Fields{"intfId": intfId})
- return nil
- }
- } else { // "single-instance"
- if tpInst, err := t.getSingleInstanceTp(ctx, tpInstPath); err != nil {
- logger.Errorw(ctx, "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
- t.AllocIDMgmtLock.Lock()
- tcontIDs, err = t.resourceMgr.GetResourceID(ctx, intfId, t.resourceMgr.GetResourceTypeAllocID(), 1)
- t.AllocIDMgmtLock.Unlock()
- if err != nil {
- logger.Errorw(ctx, "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)
- }
- }
- logger.Debugw(ctx, "Num GEM ports in TP:", log.Fields{"NumGemPorts": tp.NumGemPorts})
- t.GemPortIDMgmtLock.Lock()
- gemPorts, err = t.resourceMgr.GetResourceID(ctx, intfId, t.resourceMgr.GetResourceTypeGemPortID(), tp.NumGemPorts)
- t.GemPortIDMgmtLock.Unlock()
- if err != nil {
- logger.Errorw(ctx, "Error getting gemport ids from rsrcrMgr", log.Fields{"intfId": intfId, "numGemports": tp.NumGemPorts})
- return nil
- }
- logger.Infow(ctx, "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})
- }
-
- logger.Info(ctx, "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}
-}
-
-// allocateTPInstance function for EPON
-func (t *TechProfileMgr) allocateEponTPInstance(ctx context.Context, uniPortName string, tp *DefaultEponProfile, intfId uint32, tpInstPath string) *EponProfile {
-
- var usQueueAttributeList []iUpstreamQueueAttribute
- var dsQueueAttributeList []iDownstreamQueueAttribute
- var tcontIDs []uint32
- var gemPorts []uint32
- var err error
-
- logger.Infow(ctx, "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 {
- logger.Errorw(ctx, "Error getting alloc id from rsrcrMgr", log.Fields{"intfId": intfId})
- return nil
- }
- } else { // "single-instance"
- if tpInst, err := t.getSingleInstanceEponTp(ctx, tpInstPath); err != nil {
- logger.Errorw(ctx, "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 {
- logger.Errorw(ctx, "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.AllocID)
- }
- }
- logger.Debugw(ctx, "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 {
- logger.Errorw(ctx, "Error getting gemport ids from rsrcrMgr", log.Fields{"intfId": intfId, "numGemports": tp.NumGemPorts})
- return nil
- }
- logger.Infow(ctx, "Allocated tconts and GEM ports successfully", log.Fields{"tconts": tcontIDs, "gemports": gemPorts})
- for index := 0; index < int(tp.NumGemPorts); index++ {
- usQueueAttributeList = append(usQueueAttributeList,
- iUpstreamQueueAttribute{GemportID: gemPorts[index],
- MaxQueueSize: tp.UpstreamQueueAttributeList[index].MaxQueueSize,
- PbitMap: tp.UpstreamQueueAttributeList[index].PbitMap,
- AesEncryption: tp.UpstreamQueueAttributeList[index].AesEncryption,
- TrafficType: tp.UpstreamQueueAttributeList[index].TrafficType,
- UnsolicitedGrantSize: tp.UpstreamQueueAttributeList[index].UnsolicitedGrantSize,
- NominalInterval: tp.UpstreamQueueAttributeList[index].NominalInterval,
- ToleratedPollJitter: tp.UpstreamQueueAttributeList[index].ToleratedPollJitter,
- RequestTransmissionPolicy: tp.UpstreamQueueAttributeList[index].RequestTransmissionPolicy,
- NumQueueSet: tp.UpstreamQueueAttributeList[index].NumQueueSet,
- QThresholds: tp.UpstreamQueueAttributeList[index].QThresholds,
- SchedulingPolicy: tp.UpstreamQueueAttributeList[index].SchedulingPolicy,
- PriorityQueue: tp.UpstreamQueueAttributeList[index].PriorityQueue,
- Weight: tp.UpstreamQueueAttributeList[index].Weight,
- DiscardPolicy: tp.UpstreamQueueAttributeList[index].DiscardPolicy,
- DiscardConfig: tp.UpstreamQueueAttributeList[index].DiscardConfig})
- }
-
- logger.Info(ctx, "length of DownstreamGemPortAttributeList", len(tp.DownstreamQueueAttributeList))
- for index := 0; index < int(tp.NumGemPorts); index++ {
- dsQueueAttributeList = append(dsQueueAttributeList,
- iDownstreamQueueAttribute{GemportID: gemPorts[index],
- MaxQueueSize: tp.DownstreamQueueAttributeList[index].MaxQueueSize,
- PbitMap: tp.DownstreamQueueAttributeList[index].PbitMap,
- AesEncryption: tp.DownstreamQueueAttributeList[index].AesEncryption,
- SchedulingPolicy: tp.DownstreamQueueAttributeList[index].SchedulingPolicy,
- PriorityQueue: tp.DownstreamQueueAttributeList[index].PriorityQueue,
- Weight: tp.DownstreamQueueAttributeList[index].Weight,
- DiscardPolicy: tp.DownstreamQueueAttributeList[index].DiscardPolicy,
- DiscardConfig: tp.DownstreamQueueAttributeList[index].DiscardConfig})
- }
-
- return &EponProfile{
- SubscriberIdentifier: uniPortName,
- Name: tp.Name,
- ProfileType: tp.ProfileType,
- Version: tp.Version,
- NumGemPorts: tp.NumGemPorts,
- InstanceCtrl: tp.InstanceCtrl,
- EponAttribute: tp.EponAttribute,
- AllocID: tcontIDs[0],
- UpstreamQueueAttributeList: usQueueAttributeList,
- DownstreamQueueAttributeList: dsQueueAttributeList}
-}
-
-// 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 {
- logger.Errorw(ctx, "error-unmarshal-kv-pair", log.Fields{"keyPath": keyPath, "value": value})
- return nil, errors.New("error-unmarshal-kv-pair")
- } else {
- logger.Debugw(ctx, "found-valid-tp-instance-on-another-uni", log.Fields{"keyPath": keyPath})
- return &tpInst, nil
- }
- }
- }
- return nil, nil
-}
-
-func (t *TechProfileMgr) getSingleInstanceEponTp(ctx context.Context, tpPath string) (*EponProfile, error) {
- var tpInst EponProfile
-
- // 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 {
- logger.Errorw(ctx, "error-unmarshal-kv-pair", log.Fields{"keyPath": keyPath, "value": value})
- return nil, errors.New("error-unmarshal-kv-pair")
- } else {
- logger.Debugw(ctx, "found-valid-tp-instance-on-another-uni", log.Fields{"keyPath": keyPath})
- return &tpInst, nil
- }
- }
- }
- return nil, nil
-}
-
-func (t *TechProfileMgr) getDefaultTechProfile(ctx context.Context) *DefaultTechProfile {
- var usGemPortAttributeList []GemPortAttribute
- var dsGemPortAttributeList []GemPortAttribute
-
- for _, pbit := range t.config.DefaultPbits {
- logger.Debugw(ctx, "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}
-}
-
-// getDefaultTechProfile function for EPON
-func (t *TechProfileMgr) getDefaultEponProfile(ctx context.Context) *DefaultEponProfile {
-
- var usQueueAttributeList []UpstreamQueueAttribute
- var dsQueueAttributeList []DownstreamQueueAttribute
-
- for _, pbit := range t.config.DefaultPbits {
- logger.Debugw(ctx, "Creating Queue", log.Fields{"pbit": pbit})
- usQueueAttributeList = append(usQueueAttributeList,
- UpstreamQueueAttribute{
- MaxQueueSize: defaultMaxQueueSize,
- PbitMap: pbit,
- AesEncryption: defaultAESEncryption,
- TrafficType: defaultTrafficType,
- UnsolicitedGrantSize: defaultUnsolicitedGrantSize,
- NominalInterval: defaultNominalInterval,
- ToleratedPollJitter: defaultToleratedPollJitter,
- RequestTransmissionPolicy: defaultRequestTransmissionPolicy,
- NumQueueSet: defaultNumQueueSet,
- QThresholds: QThresholds{
- QThreshold1: defaultQThreshold1,
- QThreshold2: defaultQThreshold2,
- QThreshold3: defaultQThreshold3,
- QThreshold4: defaultQThreshold4,
- QThreshold5: defaultQThreshold5,
- QThreshold6: defaultQThreshold6,
- QThreshold7: defaultQThreshold7},
- SchedulingPolicy: SchedulingPolicy_name[defaultSchedulePolicy],
- PriorityQueue: defaultPriorityQueue,
- Weight: defaultQueueWeight,
- DiscardPolicy: DiscardPolicy_name[defaultdropPolicy],
- DiscardConfig: DiscardConfig{
- MinThreshold: defaultMinThreshold,
- MaxThreshold: defaultMaxThreshold,
- MaxProbability: defaultMaxProbability}})
- dsQueueAttributeList = append(dsQueueAttributeList,
- DownstreamQueueAttribute{
- 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}})
- }
- return &DefaultEponProfile{
- Name: t.config.DefaultTPName,
- ProfileType: t.resourceMgr.GetTechnology(),
- Version: t.config.TPVersion,
- NumGemPorts: uint32(len(usQueueAttributeList)),
- InstanceCtrl: InstanceControl{
- Onu: defaultOnuInstance,
- Uni: defaultUniInstance,
- MaxGemPayloadSize: defaultGemPayloadSize},
- EponAttribute: EponAttribute{
- PackageType: defaultPakageType},
- UpstreamQueueAttributeList: usQueueAttributeList,
- DownstreamQueueAttributeList: dsQueueAttributeList}
-}
-
-func (t *TechProfileMgr) GetprotoBufParamValue(ctx context.Context, 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 {
- logger.Debugw(ctx, "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 {
- logger.Error(ctx, "Could not find proto parameter", log.Fields{"paramType": paramType, "key": paramKey})
- return -1
- }
- logger.Debugw(ctx, "Got value in proto", log.Fields{"key": paramKey, "value": result})
- return result
-}
-
-func (t *TechProfileMgr) GetUsScheduler(ctx context.Context, tpInstance *TechProfile) (*tp_pb.SchedulerConfig, error) {
- dir := tp_pb.Direction(t.GetprotoBufParamValue(ctx, "direction", tpInstance.UsScheduler.Direction))
- if dir == -1 {
- logger.Errorf(ctx, "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(ctx, "additional_bw", tpInstance.UsScheduler.AdditionalBw))
- if bw == -1 {
- logger.Errorf(ctx, "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(ctx, "sched_policy", tpInstance.UsScheduler.QSchedPolicy))
- if policy == -1 {
- logger.Errorf(ctx, "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(ctx context.Context, tpInstance *TechProfile) (*tp_pb.SchedulerConfig, error) {
-
- dir := tp_pb.Direction(t.GetprotoBufParamValue(ctx, "direction", tpInstance.DsScheduler.Direction))
- if dir == -1 {
- logger.Errorf(ctx, "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(ctx, "additional_bw", tpInstance.DsScheduler.AdditionalBw))
- if bw == -1 {
- logger.Errorf(ctx, "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(ctx, "sched_policy", tpInstance.DsScheduler.QSchedPolicy))
- if policy == -1 {
- logger.Errorf(ctx, "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(ctx context.Context, 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(ctx, "sched_policy", tp.UpstreamGemPortAttributeList[Count].SchedulingPolicy)
- if schedPolicy == -1 {
- logger.Errorf(ctx, "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(ctx, "discard_policy", tp.UpstreamGemPortAttributeList[Count].DiscardPolicy)
- if discardPolicy == -1 {
- logger.Errorf(ctx, "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(ctx, "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),
- })
- }
- logger.Debugw(ctx, "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(ctx, "sched_policy", tp.DownstreamGemPortAttributeList[Count].SchedulingPolicy)
- if schedPolicy == -1 {
- logger.Errorf(ctx, "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(ctx, "discard_policy", tp.DownstreamGemPortAttributeList[Count].DiscardPolicy)
- if discardPolicy == -1 {
- logger.Errorf(ctx, "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(ctx, "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),
- })
- }
- logger.Debugw(ctx, "Downstream Traffic queue list ", log.Fields{"queuelist": GemPorts})
- return GemPorts, nil
- }
-
- logger.Errorf(ctx, "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(ctx context.Context, 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(ctx, "direction", tp.DsScheduler.Direction)),
- GemportId: tp.DownstreamGemPortAttributeList[Count].McastGemID,
- PbitMap: tp.DownstreamGemPortAttributeList[Count].PbitMap,
- AesEncryption: encryp,
- SchedPolicy: tp_pb.SchedulingPolicy(tpm.GetprotoBufParamValue(ctx, "sched_policy", tp.DownstreamGemPortAttributeList[Count].SchedulingPolicy)),
- Priority: tp.DownstreamGemPortAttributeList[Count].PriorityQueue,
- Weight: tp.DownstreamGemPortAttributeList[Count].Weight,
- DiscardPolicy: tp_pb.DiscardPolicy(tpm.GetprotoBufParamValue(ctx, "discard_policy", tp.DownstreamGemPortAttributeList[Count].DiscardPolicy)),
- })
- }
- logger.Debugw(ctx, "Downstream Multicast Traffic queue list ", log.Fields{"queuelist": mcastTrafficQueues})
- return mcastTrafficQueues
-}
-
-func (tpm *TechProfileMgr) GetUsTrafficScheduler(ctx context.Context, tp *TechProfile) *tp_pb.TrafficScheduler {
- UsScheduler, _ := tpm.GetUsScheduler(ctx, tp)
-
- return &tp_pb.TrafficScheduler{Direction: UsScheduler.Direction,
- AllocId: tp.UsScheduler.AllocID,
- Scheduler: UsScheduler}
-}
-
-func (t *TechProfileMgr) GetGemportIDForPbit(ctx context.Context, tp interface{}, dir tp_pb.Direction, pbit uint32) uint32 {
- /*
- Function to get the Gemport ID mapped to a pbit.
- */
- switch tp := tp.(type) {
- case *TechProfile:
- if dir == tp_pb.Direction_UPSTREAM {
- // upstream GEM ports
- numGemPorts := len(tp.UpstreamGemPortAttributeList)
- for gemCnt := 0; gemCnt < numGemPorts; gemCnt++ {
- lenOfPbitMap := len(tp.UpstreamGemPortAttributeList[gemCnt].PbitMap)
- for pbitMapIdx := 2; pbitMapIdx < lenOfPbitMap; pbitMapIdx++ {
- // Given a sample pbit map string "0b00000001", lenOfPbitMap is 10
- // "lenOfPbitMap - pbitMapIdx + 1" will give pbit-i th value from LSB position in the pbit map string
- if p, err := strconv.Atoi(string(tp.UpstreamGemPortAttributeList[gemCnt].PbitMap[lenOfPbitMap-pbitMapIdx+1])); err == nil {
- if uint32(pbitMapIdx-2) == pbit && p == 1 { // Check this p-bit is set
- logger.Debugw(ctx, "Found-US-GEMport-for-Pcp", log.Fields{"pbit": pbit, "GEMport": tp.UpstreamGemPortAttributeList[gemCnt].GemportID})
- return tp.UpstreamGemPortAttributeList[gemCnt].GemportID
- }
- }
- }
- }
- } else if dir == tp_pb.Direction_DOWNSTREAM {
- //downstream GEM ports
- numGemPorts := len(tp.DownstreamGemPortAttributeList)
- for gemCnt := 0; gemCnt < numGemPorts; gemCnt++ {
- lenOfPbitMap := len(tp.DownstreamGemPortAttributeList[gemCnt].PbitMap)
- for pbitMapIdx := 2; pbitMapIdx < lenOfPbitMap; pbitMapIdx++ {
- // Given a sample pbit map string "0b00000001", lenOfPbitMap is 10
- // "lenOfPbitMap - pbitMapIdx + 1" will give pbit-i th value from LSB position in the pbit map string
- if p, err := strconv.Atoi(string(tp.DownstreamGemPortAttributeList[gemCnt].PbitMap[lenOfPbitMap-pbitMapIdx+1])); err == nil {
- if uint32(pbitMapIdx-2) == pbit && p == 1 { // Check this p-bit is set
- logger.Debugw(ctx, "Found-DS-GEMport-for-Pcp", log.Fields{"pbit": pbit, "GEMport": tp.DownstreamGemPortAttributeList[gemCnt].GemportID})
- return tp.DownstreamGemPortAttributeList[gemCnt].GemportID
- }
- }
- }
- }
- }
- logger.Errorw(ctx, "No-GemportId-Found-For-Pcp", log.Fields{"pcpVlan": pbit})
- case *EponProfile:
- if dir == tp_pb.Direction_UPSTREAM {
- // upstream GEM ports
- numGemPorts := len(tp.UpstreamQueueAttributeList)
- for gemCnt := 0; gemCnt < numGemPorts; gemCnt++ {
- lenOfPbitMap := len(tp.UpstreamQueueAttributeList[gemCnt].PbitMap)
- for pbitMapIdx := 2; pbitMapIdx < lenOfPbitMap; pbitMapIdx++ {
- // Given a sample pbit map string "0b00000001", lenOfPbitMap is 10
- // "lenOfPbitMap - pbitMapIdx + 1" will give pbit-i th value from LSB position in the pbit map string
- if p, err := strconv.Atoi(string(tp.UpstreamQueueAttributeList[gemCnt].PbitMap[lenOfPbitMap-pbitMapIdx+1])); err == nil {
- if uint32(pbitMapIdx-2) == pbit && p == 1 { // Check this p-bit is set
- logger.Debugw(ctx, "Found-US-Queue-for-Pcp", log.Fields{"pbit": pbit, "Queue": tp.UpstreamQueueAttributeList[gemCnt].GemportID})
- return tp.UpstreamQueueAttributeList[gemCnt].GemportID
- }
- }
- }
- }
- } else if dir == tp_pb.Direction_DOWNSTREAM {
- //downstream GEM ports
- numGemPorts := len(tp.DownstreamQueueAttributeList)
- for gemCnt := 0; gemCnt < numGemPorts; gemCnt++ {
- lenOfPbitMap := len(tp.DownstreamQueueAttributeList[gemCnt].PbitMap)
- for pbitMapIdx := 2; pbitMapIdx < lenOfPbitMap; pbitMapIdx++ {
- // Given a sample pbit map string "0b00000001", lenOfPbitMap is 10
- // "lenOfPbitMap - pbitMapIdx + 1" will give pbit-i th value from LSB position in the pbit map string
- if p, err := strconv.Atoi(string(tp.DownstreamQueueAttributeList[gemCnt].PbitMap[lenOfPbitMap-pbitMapIdx+1])); err == nil {
- if uint32(pbitMapIdx-2) == pbit && p == 1 { // Check this p-bit is set
- logger.Debugw(ctx, "Found-DS-Queue-for-Pcp", log.Fields{"pbit": pbit, "Queue": tp.DownstreamQueueAttributeList[gemCnt].GemportID})
- return tp.DownstreamQueueAttributeList[gemCnt].GemportID
- }
- }
- }
- }
- }
- logger.Errorw(ctx, "No-QueueId-Found-For-Pcp", log.Fields{"pcpVlan": pbit})
- default:
- logger.Errorw(ctx, "unknown-tech", log.Fields{"tp": tp})
- }
- 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) interface{} {
- var tpTech TechProfile
- var tpEpon EponProfile
-
- 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 {
- tech := t.resourceMgr.GetTechnology()
- tpInstancesTech := make([]TechProfile, 0, len(kvPairs))
- tpInstancesEpon := make([]EponProfile, 0, len(kvPairs))
-
- for kvPath, kvPair := range kvPairs {
- if value, err := kvstore.ToByte(kvPair.Value); err == nil {
- if tech == xgspon || tech == gpon {
- if err = json.Unmarshal(value, &tpTech); err != nil {
- logger.Errorw(ctx, "error-unmarshal-kv-pair", log.Fields{"kvPath": kvPath, "value": value})
- continue
- } else {
- tpInstancesTech = append(tpInstancesTech, tpTech)
- }
- } else if tech == epon {
- if err = json.Unmarshal(value, &tpEpon); err != nil {
- logger.Errorw(ctx, "error-unmarshal-kv-pair", log.Fields{"kvPath": kvPath, "value": value})
- continue
- } else {
- tpInstancesEpon = append(tpInstancesEpon, tpEpon)
- }
- }
- }
- }
-
- switch tech {
- case xgspon, gpon:
- return tpInstancesTech
- case epon:
- return tpInstancesEpon
- default:
- logger.Errorw(ctx, "unknown-technology", log.Fields{"tech": tech})
- return nil
- }
- }
- return nil
-}
diff --git a/vendor/github.com/opencord/voltha-lib-go/v4/pkg/techprofile/tech_profile_if.go b/vendor/github.com/opencord/voltha-lib-go/v4/pkg/techprofile/tech_profile_if.go
deleted file mode 100644
index 0c5e273..0000000
--- a/vendor/github.com/opencord/voltha-lib-go/v4/pkg/techprofile/tech_profile_if.go
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- * 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/v4/pkg/db"
- tp_pb "github.com/opencord/voltha-protos/v4/go/tech_profile"
-)
-
-type TechProfileIf interface {
- SetKVClient(ctx context.Context) *db.Backend
- GetTechProfileInstanceKVPath(ctx context.Context, techProfiletblID uint32, uniPortName string) string
- GetTPInstanceFromKVStore(ctx context.Context, techProfiletblID uint32, path string) (interface{}, error)
- CreateTechProfInstance(ctx context.Context, techProfiletblID uint32, uniPortName string, intfId uint32) (interface{}, error)
- DeleteTechProfileInstance(ctx context.Context, techProfiletblID uint32, uniPortName string) error
- GetprotoBufParamValue(ctx context.Context, paramType string, paramKey string) int32
- GetUsScheduler(ctx context.Context, tpInstance *TechProfile) (*tp_pb.SchedulerConfig, error)
- GetDsScheduler(ctx context.Context, tpInstance *TechProfile) (*tp_pb.SchedulerConfig, error)
- GetTrafficScheduler(tpInstance *TechProfile, SchedCfg *tp_pb.SchedulerConfig,
- ShapingCfg *tp_pb.TrafficShapingInfo) *tp_pb.TrafficScheduler
- GetTrafficQueues(ctx context.Context, tp *TechProfile, Dir tp_pb.Direction) ([]*tp_pb.TrafficQueue, error)
- GetMulticastTrafficQueues(ctx context.Context, tp *TechProfile) []*tp_pb.TrafficQueue
- GetGemportIDForPbit(ctx context.Context, tp interface{}, Dir tp_pb.Direction, pbit uint32) uint32
- FindAllTpInstances(ctx context.Context, techProfiletblID uint32, ponIntf uint32, onuID uint32) interface{}
-}
diff --git a/vendor/github.com/opencord/voltha-lib-go/v4/pkg/db/backend.go b/vendor/github.com/opencord/voltha-lib-go/v7/pkg/db/backend.go
similarity index 92%
rename from vendor/github.com/opencord/voltha-lib-go/v4/pkg/db/backend.go
rename to vendor/github.com/opencord/voltha-lib-go/v7/pkg/db/backend.go
index d6867a5..2e57a27 100644
--- a/vendor/github.com/opencord/voltha-lib-go/v4/pkg/db/backend.go
+++ b/vendor/github.com/opencord/voltha-lib-go/v7/pkg/db/backend.go
@@ -23,8 +23,8 @@
"sync"
"time"
- "github.com/opencord/voltha-lib-go/v4/pkg/db/kvstore"
- "github.com/opencord/voltha-lib-go/v4/pkg/log"
+ "github.com/opencord/voltha-lib-go/v7/pkg/db/kvstore"
+ "github.com/opencord/voltha-lib-go/v7/pkg/log"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)
@@ -75,8 +75,6 @@
func (b *Backend) newClient(ctx context.Context, address string, timeout time.Duration) (kvstore.Client, error) {
switch b.StoreType {
- case "consul":
- return kvstore.NewConsulClient(ctx, address, timeout)
case "etcd":
return kvstore.NewEtcdClient(ctx, address, timeout, log.WarnLevel)
}
@@ -170,9 +168,6 @@
case codes.DataLoss:
alive = false
}
-
- //} else {
- // TODO: Implement for consul backend; would it be needed ever?
}
}
@@ -239,6 +234,21 @@
return err
}
+// DeleteWithPrefix removes items having prefix key
+func (b *Backend) DeleteWithPrefix(ctx context.Context, prefixKey string) error {
+ span, ctx := log.CreateChildSpan(ctx, "etcd-delete-with-prefix")
+ defer span.Finish()
+
+ formattedPath := b.makePath(ctx, prefixKey)
+ logger.Debugw(ctx, "deleting-prefix-key", log.Fields{"key": prefixKey, "path": formattedPath})
+
+ err := b.Client.DeleteWithPrefix(ctx, formattedPath)
+
+ b.updateLiveness(ctx, b.isErrorIndicatingAliveKvstore(ctx, 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 {
span, ctx := log.CreateChildSpan(ctx, "etcd-create-watch")
diff --git a/vendor/github.com/opencord/voltha-lib-go/v4/pkg/db/common.go b/vendor/github.com/opencord/voltha-lib-go/v7/pkg/db/common.go
similarity index 94%
rename from vendor/github.com/opencord/voltha-lib-go/v4/pkg/db/common.go
rename to vendor/github.com/opencord/voltha-lib-go/v7/pkg/db/common.go
index 25cddf5..d8a0571 100644
--- a/vendor/github.com/opencord/voltha-lib-go/v4/pkg/db/common.go
+++ b/vendor/github.com/opencord/voltha-lib-go/v7/pkg/db/common.go
@@ -16,7 +16,7 @@
package db
import (
- "github.com/opencord/voltha-lib-go/v4/pkg/log"
+ "github.com/opencord/voltha-lib-go/v7/pkg/log"
)
var logger log.CLogger
diff --git a/vendor/github.com/opencord/voltha-lib-go/v4/pkg/db/kvstore/client.go b/vendor/github.com/opencord/voltha-lib-go/v7/pkg/db/kvstore/client.go
similarity index 92%
rename from vendor/github.com/opencord/voltha-lib-go/v4/pkg/db/kvstore/client.go
rename to vendor/github.com/opencord/voltha-lib-go/v7/pkg/db/kvstore/client.go
index 480d476..e4b1fff 100644
--- a/vendor/github.com/opencord/voltha-lib-go/v4/pkg/db/kvstore/client.go
+++ b/vendor/github.com/opencord/voltha-lib-go/v7/pkg/db/kvstore/client.go
@@ -21,8 +21,6 @@
)
const (
- // Default timeout in seconds when making a kvstore request
- defaultKVGetTimeout = 5 * time.Second
// Maximum channel buffer between publisher/subscriber goroutines
maxClientChannelBufferSize = 10
)
@@ -80,14 +78,18 @@
Get(ctx context.Context, key string) (*KVPair, error)
Put(ctx context.Context, key string, value interface{}) error
Delete(ctx context.Context, key string) error
+ DeleteWithPrefix(ctx context.Context, prefixKey string) error
+ Watch(ctx context.Context, key string, withPrefix bool) chan *Event
+ IsConnectionUp(ctx context.Context) bool // timeout in second
+ CloseWatch(ctx context.Context, key string, ch chan *Event)
+ Close(ctx context.Context)
+
+ // These APIs are not used. They will be cleaned up in release Voltha 2.9.
+ // It's not cleaned now to limit changes in all components
Reserve(ctx context.Context, key string, value interface{}, ttl time.Duration) (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 time.Duration) error
ReleaseLock(lockName string) error
- IsConnectionUp(ctx context.Context) bool // timeout in second
- CloseWatch(ctx context.Context, key string, ch chan *Event)
- Close(ctx context.Context)
}
diff --git a/vendor/github.com/opencord/voltha-lib-go/v4/pkg/db/kvstore/common.go b/vendor/github.com/opencord/voltha-lib-go/v7/pkg/db/kvstore/common.go
similarity index 94%
rename from vendor/github.com/opencord/voltha-lib-go/v4/pkg/db/kvstore/common.go
rename to vendor/github.com/opencord/voltha-lib-go/v7/pkg/db/kvstore/common.go
index 99c603d..8ac2a4a 100644
--- a/vendor/github.com/opencord/voltha-lib-go/v4/pkg/db/kvstore/common.go
+++ b/vendor/github.com/opencord/voltha-lib-go/v7/pkg/db/kvstore/common.go
@@ -16,7 +16,7 @@
package kvstore
import (
- "github.com/opencord/voltha-lib-go/v4/pkg/log"
+ "github.com/opencord/voltha-lib-go/v7/pkg/log"
)
var logger log.CLogger
diff --git a/vendor/github.com/opencord/voltha-lib-go/v7/pkg/db/kvstore/etcdclient.go b/vendor/github.com/opencord/voltha-lib-go/v7/pkg/db/kvstore/etcdclient.go
new file mode 100644
index 0000000..6ca5329
--- /dev/null
+++ b/vendor/github.com/opencord/voltha-lib-go/v7/pkg/db/kvstore/etcdclient.go
@@ -0,0 +1,474 @@
+/*
+ * 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"
+ "os"
+ "strconv"
+ "sync"
+ "time"
+
+ "github.com/opencord/voltha-lib-go/v7/pkg/log"
+ v3Client "go.etcd.io/etcd/clientv3"
+ v3rpcTypes "go.etcd.io/etcd/etcdserver/api/v3rpc/rpctypes"
+)
+
+const (
+ poolCapacityEnvName = "VOLTHA_ETCD_CLIENT_POOL_CAPACITY"
+ maxUsageEnvName = "VOLTHA_ETCD_CLIENT_MAX_USAGE"
+)
+
+const (
+ defaultMaxPoolCapacity = 1000 // Default size of an Etcd Client pool
+ defaultMaxPoolUsage = 100 // Maximum concurrent request an Etcd Client is allowed to process
+)
+
+// EtcdClient represents the Etcd KV store client
+type EtcdClient struct {
+ pool EtcdClientAllocator
+ watchedChannels sync.Map
+ watchedClients map[string]*v3Client.Client
+ watchedClientsLock sync.RWMutex
+}
+
+// NewEtcdCustomClient returns a new client for the Etcd KV store allowing
+// the called to specify etcd client configuration
+func NewEtcdCustomClient(ctx context.Context, addr string, timeout time.Duration, level log.LogLevel) (*EtcdClient, error) {
+ // Get the capacity and max usage from the environment
+ capacity := defaultMaxPoolCapacity
+ maxUsage := defaultMaxPoolUsage
+ if capacityStr, present := os.LookupEnv(poolCapacityEnvName); present {
+ if val, err := strconv.Atoi(capacityStr); err == nil {
+ capacity = val
+ logger.Infow(ctx, "env-variable-set", log.Fields{"pool-capacity": capacity})
+ } else {
+ logger.Warnw(ctx, "invalid-capacity-value", log.Fields{"error": err, "capacity": capacityStr})
+ }
+ }
+ if maxUsageStr, present := os.LookupEnv(maxUsageEnvName); present {
+ if val, err := strconv.Atoi(maxUsageStr); err == nil {
+ maxUsage = val
+ logger.Infow(ctx, "env-variable-set", log.Fields{"max-usage": maxUsage})
+ } else {
+ logger.Warnw(ctx, "invalid-max-usage-value", log.Fields{"error": err, "max-usage": maxUsageStr})
+ }
+ }
+
+ var err error
+
+ pool, err := NewRoundRobinEtcdClientAllocator([]string{addr}, timeout, capacity, maxUsage, level)
+ if err != nil {
+ logger.Errorw(ctx, "failed-to-create-rr-client", log.Fields{
+ "error": err,
+ })
+ }
+
+ logger.Infow(ctx, "etcd-pool-created", log.Fields{"capacity": capacity, "max-usage": maxUsage})
+
+ return &EtcdClient{pool: pool,
+ watchedClients: make(map[string]*v3Client.Client),
+ }, nil
+}
+
+// NewEtcdClient returns a new client for the Etcd KV store
+func NewEtcdClient(ctx context.Context, addr string, timeout time.Duration, level log.LogLevel) (*EtcdClient, error) {
+ return NewEtcdCustomClient(ctx, addr, timeout, level)
+}
+
+// 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
+ }
+ 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) {
+ client, err := c.pool.Get(ctx)
+ if err != nil {
+ return nil, err
+ }
+ defer c.pool.Put(client)
+ resp, err := client.Get(ctx, key, v3Client.WithPrefix())
+
+ if err != nil {
+ logger.Error(ctx, 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) {
+ client, err := c.pool.Get(ctx)
+ if err != nil {
+ return nil, err
+ }
+ defer c.pool.Put(client)
+
+ attempt := 0
+
+startLoop:
+ for {
+ resp, err := client.Get(ctx, key)
+ if err != nil {
+ switch err {
+ case context.Canceled:
+ logger.Warnw(ctx, "context-cancelled", log.Fields{"error": err})
+ case context.DeadlineExceeded:
+ logger.Warnw(ctx, "context-deadline-exceeded", log.Fields{"error": err, "context": ctx})
+ case v3rpcTypes.ErrEmptyKey:
+ logger.Warnw(ctx, "etcd-client-error", log.Fields{"error": err})
+ case v3rpcTypes.ErrLeaderChanged,
+ v3rpcTypes.ErrGRPCNoLeader,
+ v3rpcTypes.ErrTimeout,
+ v3rpcTypes.ErrTimeoutDueToLeaderFail,
+ v3rpcTypes.ErrTimeoutDueToConnectionLost:
+ // Retry for these server errors
+ attempt += 1
+ if er := backoff(ctx, attempt); er != nil {
+ logger.Warnw(ctx, "get-retries-failed", log.Fields{"key": key, "error": er, "attempt": attempt})
+ return nil, err
+ }
+ logger.Warnw(ctx, "retrying-get", log.Fields{"key": key, "error": err, "attempt": attempt})
+ goto startLoop
+ default:
+ logger.Warnw(ctx, "etcd-server-error", log.Fields{"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 err error
+ if val, err = ToString(value); err != nil {
+ return fmt.Errorf("unexpected-type-%T", value)
+ }
+
+ client, err := c.pool.Get(ctx)
+ if err != nil {
+ return err
+ }
+ defer c.pool.Put(client)
+
+ attempt := 0
+startLoop:
+ for {
+ _, err = client.Put(ctx, key, val)
+ if err != nil {
+ switch err {
+ case context.Canceled:
+ logger.Warnw(ctx, "context-cancelled", log.Fields{"error": err})
+ case context.DeadlineExceeded:
+ logger.Warnw(ctx, "context-deadline-exceeded", log.Fields{"error": err, "context": ctx})
+ case v3rpcTypes.ErrEmptyKey:
+ logger.Warnw(ctx, "etcd-client-error", log.Fields{"error": err})
+ case v3rpcTypes.ErrLeaderChanged,
+ v3rpcTypes.ErrGRPCNoLeader,
+ v3rpcTypes.ErrTimeout,
+ v3rpcTypes.ErrTimeoutDueToLeaderFail,
+ v3rpcTypes.ErrTimeoutDueToConnectionLost:
+ // Retry for these server errors
+ attempt += 1
+ if er := backoff(ctx, attempt); er != nil {
+ logger.Warnw(ctx, "put-retries-failed", log.Fields{"key": key, "error": er, "attempt": attempt})
+ return err
+ }
+ logger.Warnw(ctx, "retrying-put", log.Fields{"key": key, "error": err, "attempt": attempt})
+ goto startLoop
+ default:
+ logger.Warnw(ctx, "etcd-server-error", 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 {
+ client, err := c.pool.Get(ctx)
+ if err != nil {
+ return err
+ }
+ defer c.pool.Put(client)
+
+ attempt := 0
+startLoop:
+ for {
+ _, err = client.Delete(ctx, key)
+ if err != nil {
+ switch err {
+ case context.Canceled:
+ logger.Warnw(ctx, "context-cancelled", log.Fields{"error": err})
+ case context.DeadlineExceeded:
+ logger.Warnw(ctx, "context-deadline-exceeded", log.Fields{"error": err, "context": ctx})
+ case v3rpcTypes.ErrEmptyKey:
+ logger.Warnw(ctx, "etcd-client-error", log.Fields{"error": err})
+ case v3rpcTypes.ErrLeaderChanged,
+ v3rpcTypes.ErrGRPCNoLeader,
+ v3rpcTypes.ErrTimeout,
+ v3rpcTypes.ErrTimeoutDueToLeaderFail,
+ v3rpcTypes.ErrTimeoutDueToConnectionLost:
+ // Retry for these server errors
+ attempt += 1
+ if er := backoff(ctx, attempt); er != nil {
+ logger.Warnw(ctx, "delete-retries-failed", log.Fields{"key": key, "error": er, "attempt": attempt})
+ return err
+ }
+ logger.Warnw(ctx, "retrying-delete", log.Fields{"key": key, "error": err, "attempt": attempt})
+ goto startLoop
+ default:
+ logger.Warnw(ctx, "etcd-server-error", log.Fields{"error": err})
+ }
+ return err
+ }
+ logger.Debugw(ctx, "key(s)-deleted", log.Fields{"key": key})
+ return nil
+ }
+}
+
+func (c *EtcdClient) DeleteWithPrefix(ctx context.Context, prefixKey string) error {
+
+ client, err := c.pool.Get(ctx)
+ if err != nil {
+ return err
+ }
+ defer c.pool.Put(client)
+
+ //delete the prefix
+ if _, err := client.Delete(ctx, prefixKey, v3Client.WithPrefix()); err != nil {
+ logger.Errorw(ctx, "failed-to-delete-prefix-key", log.Fields{"key": prefixKey, "error": err})
+ return err
+ }
+ logger.Debugw(ctx, "key(s)-deleted", log.Fields{"key": prefixKey})
+ 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 {
+ var err error
+ // Reuse the Etcd client when multiple callees are watching the same key.
+ c.watchedClientsLock.Lock()
+ client, exist := c.watchedClients[key]
+ if !exist {
+ client, err = c.pool.Get(ctx)
+ if err != nil {
+ logger.Errorw(ctx, "failed-to-an-etcd-client", log.Fields{"key": key, "error": err})
+ c.watchedClientsLock.Unlock()
+ return nil
+ }
+ c.watchedClients[key] = client
+ }
+ c.watchedClientsLock.Unlock()
+
+ w := v3Client.NewWatcher(client)
+ 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(ctx, "watched-channels", log.Fields{"len": len(channelMaps)})
+ // Launch a go routine to listen for updates
+ go c.listenForKeyChange(ctx, 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(ctx context.Context, 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(ctx, "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(ctx, "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(ctx, "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)
+ }
+
+ // If we don't have any keys being watched then return the Etcd client to the pool
+ if len(channelMaps) == 0 {
+ c.watchedClientsLock.Lock()
+ // Sanity
+ if client, ok := c.watchedClients[key]; ok {
+ c.pool.Put(client)
+ delete(c.watchedClients, key)
+ }
+ c.watchedClientsLock.Unlock()
+ }
+ logger.Infow(ctx, "watcher-channel-exiting", log.Fields{"key": key, "channel": channelMaps})
+}
+
+func (c *EtcdClient) listenForKeyChange(ctx context.Context, channel v3Client.WatchChan, ch chan<- *Event, cancel context.CancelFunc) {
+ logger.Debug(ctx, "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(ctx, "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 all the connection in the pool store client
+func (c *EtcdClient) Close(ctx context.Context) {
+ logger.Debug(ctx, "closing-etcd-pool")
+ c.pool.Close(ctx)
+}
+
+// The APIs below are not used
+var errUnimplemented = errors.New("deprecated")
+
+// Reserve is deprecated
+func (c *EtcdClient) Reserve(ctx context.Context, key string, value interface{}, ttl time.Duration) (interface{}, error) {
+ return nil, errUnimplemented
+}
+
+// ReleaseAllReservations is deprecated
+func (c *EtcdClient) ReleaseAllReservations(ctx context.Context) error {
+ return errUnimplemented
+}
+
+// ReleaseReservation is deprecated
+func (c *EtcdClient) ReleaseReservation(ctx context.Context, key string) error {
+ return errUnimplemented
+}
+
+// RenewReservation is deprecated
+func (c *EtcdClient) RenewReservation(ctx context.Context, key string) error {
+ return errUnimplemented
+}
+
+// AcquireLock is deprecated
+func (c *EtcdClient) AcquireLock(ctx context.Context, lockName string, timeout time.Duration) error {
+ return errUnimplemented
+}
+
+// ReleaseLock is deprecated
+func (c *EtcdClient) ReleaseLock(lockName string) error {
+ return errUnimplemented
+}
diff --git a/vendor/github.com/opencord/voltha-lib-go/v7/pkg/db/kvstore/etcdpool.go b/vendor/github.com/opencord/voltha-lib-go/v7/pkg/db/kvstore/etcdpool.go
new file mode 100644
index 0000000..4d33c27
--- /dev/null
+++ b/vendor/github.com/opencord/voltha-lib-go/v7/pkg/db/kvstore/etcdpool.go
@@ -0,0 +1,240 @@
+/*
+ * Copyright 2021-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 (
+ "container/list"
+ "context"
+ "errors"
+ "sync"
+ "time"
+
+ "github.com/opencord/voltha-lib-go/v7/pkg/log"
+ "go.etcd.io/etcd/clientv3"
+)
+
+// EtcdClientAllocator represents a generic interface to allocate an Etcd Client
+type EtcdClientAllocator interface {
+ Get(context.Context) (*clientv3.Client, error)
+ Put(*clientv3.Client)
+ Close(ctx context.Context)
+}
+
+// NewRoundRobinEtcdClientAllocator creates a new ETCD Client Allocator using a Round Robin scheme
+func NewRoundRobinEtcdClientAllocator(endpoints []string, timeout time.Duration, capacity, maxUsage int, level log.LogLevel) (EtcdClientAllocator, error) {
+ return &roundRobin{
+ all: make(map[*clientv3.Client]*rrEntry),
+ full: make(map[*clientv3.Client]*rrEntry),
+ waitList: list.New(),
+ max: maxUsage,
+ capacity: capacity,
+ timeout: timeout,
+ endpoints: endpoints,
+ logLevel: level,
+ closingCh: make(chan struct{}, capacity*maxUsage),
+ stopCh: make(chan struct{}),
+ }, nil
+}
+
+type rrEntry struct {
+ client *clientv3.Client
+ count int
+ age time.Time
+}
+
+type roundRobin struct {
+ //block chan struct{}
+ sync.Mutex
+ available []*rrEntry
+ all map[*clientv3.Client]*rrEntry
+ full map[*clientv3.Client]*rrEntry
+ waitList *list.List
+ max int
+ capacity int
+ timeout time.Duration
+ //ageOut time.Duration
+ endpoints []string
+ size int
+ logLevel log.LogLevel
+ closing bool
+ closingCh chan struct{}
+ stopCh chan struct{}
+}
+
+// Get returns an Etcd client. If not is available, it will create one
+// until the maximum allowed capacity. If maximum capacity has been
+// reached then it will wait until s used one is freed.
+func (r *roundRobin) Get(ctx context.Context) (*clientv3.Client, error) {
+ r.Lock()
+
+ if r.closing {
+ r.Unlock()
+ return nil, errors.New("pool-is-closing")
+ }
+
+ // first determine if we need to block, which would mean the
+ // available queue is empty and we are at capacity
+ if len(r.available) == 0 && r.size >= r.capacity {
+
+ // create a channel on which to wait and
+ // add it to the list
+ ch := make(chan struct{})
+ element := r.waitList.PushBack(ch)
+ r.Unlock()
+
+ // block until it is our turn or context
+ // expires or is canceled
+ select {
+ case <-r.stopCh:
+ logger.Info(ctx, "stop-waiting-pool-is-closing")
+ r.waitList.Remove(element)
+ return nil, errors.New("stop-waiting-pool-is-closing")
+ case <-ch:
+ r.waitList.Remove(element)
+ case <-ctx.Done():
+ r.waitList.Remove(element)
+ return nil, ctx.Err()
+ }
+ r.Lock()
+ }
+
+ defer r.Unlock()
+ if len(r.available) > 0 {
+ // pull off back end as it is operationally quicker
+ last := len(r.available) - 1
+ entry := r.available[last]
+ entry.count++
+ if entry.count >= r.max {
+ r.available = r.available[:last]
+ r.full[entry.client] = entry
+ }
+ entry.age = time.Now()
+ return entry.client, nil
+ }
+
+ logConfig := log.ConstructZapConfig(log.JSON, r.logLevel, log.Fields{})
+ // increase capacity
+ client, err := clientv3.New(clientv3.Config{
+ Endpoints: r.endpoints,
+ DialTimeout: r.timeout,
+ LogConfig: &logConfig,
+ })
+ if err != nil {
+ return nil, err
+ }
+ entry := &rrEntry{
+ client: client,
+ count: 1,
+ }
+ r.all[entry.client] = entry
+
+ if r.max > 1 {
+ r.available = append(r.available, entry)
+ } else {
+ r.full[entry.client] = entry
+ }
+ r.size++
+ return client, nil
+}
+
+// Put returns the Etcd Client back to the pool
+func (r *roundRobin) Put(client *clientv3.Client) {
+ r.Lock()
+
+ entry := r.all[client]
+ entry.count--
+
+ if r.closing {
+ // Close client if count is 0
+ if entry.count == 0 {
+ if err := entry.client.Close(); err != nil {
+ logger.Warnw(context.Background(), "error-closing-client", log.Fields{"error": err})
+ }
+ delete(r.all, entry.client)
+ }
+ // Notify Close function that a client was returned to the pool
+ r.closingCh <- struct{}{}
+ r.Unlock()
+ return
+ }
+
+ // This entry is now available for use, so
+ // if in full map add it to available and
+ // remove from full
+ if _, ok := r.full[client]; ok {
+ r.available = append(r.available, entry)
+ delete(r.full, client)
+ }
+
+ front := r.waitList.Front()
+ if front != nil {
+ ch := r.waitList.Remove(front)
+ r.Unlock()
+ // need to unblock if someone is waiting
+ ch.(chan struct{}) <- struct{}{}
+ return
+ }
+ r.Unlock()
+}
+
+func (r *roundRobin) Close(ctx context.Context) {
+ r.Lock()
+ r.closing = true
+
+ // Notify anyone waiting for a client to stop waiting
+ close(r.stopCh)
+
+ // Clean-up unused clients
+ for i := 0; i < len(r.available); i++ {
+ // Count 0 means no one is using that client
+ if r.available[i].count == 0 {
+ if err := r.available[i].client.Close(); err != nil {
+ logger.Warnw(ctx, "failure-closing-client", log.Fields{"client": r.available[i].client, "error": err})
+ }
+ // Remove client for all list
+ delete(r.all, r.available[i].client)
+ }
+ }
+
+ // Figure out how many clients are in use
+ numberInUse := 0
+ for _, rrEntry := range r.all {
+ numberInUse += rrEntry.count
+ }
+ r.Unlock()
+
+ if numberInUse == 0 {
+ logger.Info(ctx, "no-connection-in-use")
+ return
+ }
+
+ logger.Infow(ctx, "waiting-for-clients-return", log.Fields{"count": numberInUse})
+
+ // Wait for notifications when a client is returned to the pool
+ for {
+ select {
+ case <-r.closingCh:
+ numberInUse--
+ if numberInUse == 0 {
+ logger.Info(ctx, "all-connections-closed")
+ return
+ }
+ case <-ctx.Done():
+ logger.Warnw(ctx, "context-done", log.Fields{"error": ctx.Err()})
+ return
+ }
+ }
+}
diff --git a/vendor/github.com/opencord/voltha-lib-go/v4/pkg/db/kvstore/kvutils.go b/vendor/github.com/opencord/voltha-lib-go/v7/pkg/db/kvstore/kvutils.go
similarity index 64%
rename from vendor/github.com/opencord/voltha-lib-go/v4/pkg/db/kvstore/kvutils.go
rename to vendor/github.com/opencord/voltha-lib-go/v7/pkg/db/kvstore/kvutils.go
index 64e7d30..ca57542 100644
--- a/vendor/github.com/opencord/voltha-lib-go/v4/pkg/db/kvstore/kvutils.go
+++ b/vendor/github.com/opencord/voltha-lib-go/v7/pkg/db/kvstore/kvutils.go
@@ -15,7 +15,20 @@
*/
package kvstore
-import "fmt"
+import (
+ "context"
+ "fmt"
+ "math"
+ "math/rand"
+ "time"
+)
+
+const (
+ minRetryInterval = 100
+ maxRetryInterval = 5000
+ incrementalFactor = 1.2
+ jitter = 0.2
+)
// ToString converts an interface value to a string. The interface should either be of
// a string type or []byte. Otherwise, an error is returned.
@@ -42,3 +55,24 @@
return nil, fmt.Errorf("unexpected-type-%T", t)
}
}
+
+// backoff waits an amount of time that is proportional to the attempt value. The wait time in a range of
+// minRetryInterval and maxRetryInterval.
+func backoff(ctx context.Context, attempt int) error {
+ if attempt == 0 {
+ return nil
+ }
+ backoff := int(minRetryInterval + incrementalFactor*math.Exp(float64(attempt)))
+ backoff *= 1 + int(jitter*(rand.Float64()*2-1))
+ if backoff > maxRetryInterval {
+ backoff = maxRetryInterval
+ }
+ ticker := time.NewTicker(time.Duration(backoff) * time.Millisecond)
+ defer ticker.Stop()
+ select {
+ case <-ctx.Done():
+ return ctx.Err()
+ case <-ticker.C:
+ }
+ return nil
+}
diff --git a/vendor/github.com/opencord/voltha-lib-go/v4/pkg/log/common.go b/vendor/github.com/opencord/voltha-lib-go/v7/pkg/log/common.go
similarity index 100%
rename from vendor/github.com/opencord/voltha-lib-go/v4/pkg/log/common.go
rename to vendor/github.com/opencord/voltha-lib-go/v7/pkg/log/common.go
diff --git a/vendor/github.com/opencord/voltha-lib-go/v4/pkg/log/log.go b/vendor/github.com/opencord/voltha-lib-go/v7/pkg/log/log.go
similarity index 99%
rename from vendor/github.com/opencord/voltha-lib-go/v4/pkg/log/log.go
rename to vendor/github.com/opencord/voltha-lib-go/v7/pkg/log/log.go
index b8d498c..7b1a123 100644
--- a/vendor/github.com/opencord/voltha-lib-go/v4/pkg/log/log.go
+++ b/vendor/github.com/opencord/voltha-lib-go/v7/pkg/log/log.go
@@ -54,11 +54,12 @@
"context"
"errors"
"fmt"
- zp "go.uber.org/zap"
- zc "go.uber.org/zap/zapcore"
"path"
"runtime"
"strings"
+
+ zp "go.uber.org/zap"
+ zc "go.uber.org/zap/zapcore"
)
type LogLevel int8
@@ -217,7 +218,7 @@
case FatalLevel:
return "FATAL", nil
}
- return "", errors.New("Given LogLevel is invalid " + string(l))
+ return "", fmt.Errorf("Given LogLevel is invalid %d", l)
}
func getDefaultConfig(outputType string, level LogLevel, defaultFields Fields) zp.Config {
diff --git a/vendor/github.com/opencord/voltha-lib-go/v4/pkg/log/utils.go b/vendor/github.com/opencord/voltha-lib-go/v7/pkg/log/utils.go
similarity index 100%
rename from vendor/github.com/opencord/voltha-lib-go/v4/pkg/log/utils.go
rename to vendor/github.com/opencord/voltha-lib-go/v7/pkg/log/utils.go
diff --git a/vendor/github.com/opencord/voltha-lib-go/v4/pkg/ponresourcemanager/common.go b/vendor/github.com/opencord/voltha-lib-go/v7/pkg/ponresourcemanager/common.go
similarity index 94%
rename from vendor/github.com/opencord/voltha-lib-go/v4/pkg/ponresourcemanager/common.go
rename to vendor/github.com/opencord/voltha-lib-go/v7/pkg/ponresourcemanager/common.go
index 1c9a5b1..b311bb3 100644
--- a/vendor/github.com/opencord/voltha-lib-go/v4/pkg/ponresourcemanager/common.go
+++ b/vendor/github.com/opencord/voltha-lib-go/v7/pkg/ponresourcemanager/common.go
@@ -16,7 +16,7 @@
package ponresourcemanager
import (
- "github.com/opencord/voltha-lib-go/v4/pkg/log"
+ "github.com/opencord/voltha-lib-go/v7/pkg/log"
)
var logger log.CLogger
diff --git a/vendor/github.com/opencord/voltha-lib-go/v4/pkg/ponresourcemanager/ponresourcemanager.go b/vendor/github.com/opencord/voltha-lib-go/v7/pkg/ponresourcemanager/ponresourcemanager.go
similarity index 87%
rename from vendor/github.com/opencord/voltha-lib-go/v4/pkg/ponresourcemanager/ponresourcemanager.go
rename to vendor/github.com/opencord/voltha-lib-go/v7/pkg/ponresourcemanager/ponresourcemanager.go
index bbfebe3..ad6c111 100644
--- a/vendor/github.com/opencord/voltha-lib-go/v4/pkg/ponresourcemanager/ponresourcemanager.go
+++ b/vendor/github.com/opencord/voltha-lib-go/v7/pkg/ponresourcemanager/ponresourcemanager.go
@@ -25,10 +25,9 @@
"time"
bitmap "github.com/boljen/go-bitmap"
- "github.com/opencord/voltha-lib-go/v4/pkg/db"
- "github.com/opencord/voltha-lib-go/v4/pkg/db/kvstore"
- "github.com/opencord/voltha-lib-go/v4/pkg/log"
- tp "github.com/opencord/voltha-lib-go/v4/pkg/techprofile"
+ "github.com/opencord/voltha-lib-go/v7/pkg/db"
+ "github.com/opencord/voltha-lib-go/v7/pkg/db/kvstore"
+ "github.com/opencord/voltha-lib-go/v7/pkg/log"
)
const (
@@ -39,9 +38,10 @@
GEMPORT_ID = "GEMPORT_ID"
FLOW_ID = "FLOW_ID"
- //Constants for passing command line arugments
+ //Constants for passing command line arguments
OLT_MODEL_ARG = "--olt_model"
- PATH_PREFIX = "service/voltha/resource_manager/{%s}"
+
+ PATH_PREFIX = "%s/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
@@ -51,7 +51,8 @@
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"
+
+ PATH_PREFIX_FOR_CONFIG = "%s/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.
@@ -110,12 +111,11 @@
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)>
+ FLOW_ID_INFO_PATH_PREFIX = "{%s}/flow_id_info"
+ //Format: <device_id>/flow_id_info/<(pon_intf_id, onu_id)>
+ FLOW_ID_INFO_PATH_INTF_ONU_PREFIX = "{%s}/flow_id_info/{%s}"
+ //Format: <device_id>/flow_id_info/<(pon_intf_id, onu_id)><flow_id>
+ FLOW_ID_INFO_PATH = FLOW_ID_INFO_PATH_PREFIX + "/{%s}/{%d}"
//Constants for internal usage.
PON_INTF_ID = "pon_intf_id"
@@ -138,12 +138,11 @@
Technology string
DeviceType string
DeviceID string
- Backend string // ETCD, or consul
+ Backend string // ETCD only currently
Address string // address of 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.
@@ -157,15 +156,13 @@
func newKVClient(ctx context.Context, storeType string, address string, timeout time.Duration) (kvstore.Client, error) {
logger.Infow(ctx, "kv-store-type", log.Fields{"store": storeType})
switch storeType {
- case "consul":
- return kvstore.NewConsulClient(ctx, address, timeout)
case "etcd":
return kvstore.NewEtcdClient(ctx, address, timeout, log.WarnLevel)
}
return nil, errors.New("unsupported-kv-store")
}
-func SetKVClient(ctx context.Context, Technology string, Backend string, Addr string, configClient bool) *db.Backend {
+func SetKVClient(ctx context.Context, Technology string, Backend string, Addr string, configClient bool, basePathKvStore string) *db.Backend {
// 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(ctx, Backend, Addr, KVSTORE_RETRY_TIMEOUT)
@@ -176,9 +173,9 @@
var pathPrefix string
if configClient {
- pathPrefix = PATH_PREFIX_FOR_CONFIG
+ pathPrefix = fmt.Sprintf(PATH_PREFIX_FOR_CONFIG, basePathKvStore)
} else {
- pathPrefix = fmt.Sprintf(PATH_PREFIX, Technology)
+ pathPrefix = fmt.Sprintf(PATH_PREFIX, basePathKvStore, Technology)
}
kvbackend := &db.Backend{
@@ -192,29 +189,25 @@
}
// NewPONResourceManager creates a new PON resource manager.
-func NewPONResourceManager(ctx context.Context, Technology string, DeviceType string, DeviceID string, Backend string, Address string) (*PONResourceManager, error) {
+func NewPONResourceManager(ctx context.Context, Technology string, DeviceType string, DeviceID string, Backend string, Address string, basePathKvStore string) (*PONResourceManager, error) {
var PONMgr PONResourceManager
PONMgr.Technology = Technology
PONMgr.DeviceType = DeviceType
PONMgr.DeviceID = DeviceID
PONMgr.Backend = Backend
PONMgr.Address = Address
- PONMgr.KVStore = SetKVClient(ctx, Technology, Backend, Address, false)
+ PONMgr.KVStore = SetKVClient(ctx, Technology, Backend, Address, false, basePathKvStore)
if PONMgr.KVStore == nil {
logger.Error(ctx, "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(ctx, Technology, Backend, Address, true)
+ PONMgr.KVStoreForConfig = SetKVClient(ctx, Technology, Backend, Address, true, basePathKvStore)
if PONMgr.KVStoreForConfig == nil {
logger.Error(ctx, "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(ctx, &PONMgr, Backend, Address); PONMgr.TechProfileMgr == nil {
- logger.Error(ctx, "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)
@@ -425,6 +418,43 @@
return err
}
+func (PONRMgr *PONResourceManager) InitDeviceResourcePoolForIntf(ctx context.Context, intfID uint32) error {
+
+ logger.Debug(ctx, "Init resource ranges for intf %d", intfID)
+
+ var err error
+
+ if err = PONRMgr.InitResourceIDPool(ctx, intfID, ONU_ID,
+ PONRMgr.PonResourceRanges[ONU_ID_START_IDX].(uint32),
+ PONRMgr.PonResourceRanges[ONU_ID_END_IDX].(uint32)); err != nil {
+ logger.Error(ctx, "Failed to init ONU ID resource pool")
+ return err
+ }
+
+ if err = PONRMgr.InitResourceIDPool(ctx, intfID, ALLOC_ID,
+ PONRMgr.PonResourceRanges[ALLOC_ID_START_IDX].(uint32),
+ PONRMgr.PonResourceRanges[ALLOC_ID_END_IDX].(uint32)); err != nil {
+ logger.Error(ctx, "Failed to init ALLOC ID resource pool ")
+ return err
+ }
+
+ if err = PONRMgr.InitResourceIDPool(ctx, intfID, GEMPORT_ID,
+ PONRMgr.PonResourceRanges[GEMPORT_ID_START_IDX].(uint32),
+ PONRMgr.PonResourceRanges[GEMPORT_ID_END_IDX].(uint32)); err != nil {
+ logger.Error(ctx, "Failed to init GEMPORT ID resource pool")
+ return err
+ }
+
+ if err = PONRMgr.InitResourceIDPool(ctx, intfID, FLOW_ID,
+ PONRMgr.PonResourceRanges[FLOW_ID_START_IDX].(uint32),
+ PONRMgr.PonResourceRanges[FLOW_ID_END_IDX].(uint32)); err != nil {
+ logger.Error(ctx, "Failed to init FLOW ID resource pool")
+ return err
+ }
+
+ return nil
+}
+
func (PONRMgr *PONResourceManager) ClearDeviceResourcePool(ctx context.Context) error {
//Clear resource pool for all PON ports.
@@ -488,6 +518,33 @@
return nil
}
+func (PONRMgr *PONResourceManager) ClearDeviceResourcePoolForIntf(ctx context.Context, intfID uint32) error {
+
+ logger.Debugf(ctx, "Clear resource ranges for intf %d", intfID)
+
+ if status := PONRMgr.ClearResourceIDPool(ctx, intfID, ONU_ID); !status {
+ logger.Error(ctx, "Failed to clear ONU ID resource pool")
+ return errors.New("Failed to clear ONU ID resource pool")
+ }
+
+ if status := PONRMgr.ClearResourceIDPool(ctx, intfID, ALLOC_ID); !status {
+ logger.Error(ctx, "Failed to clear ALLOC ID resource pool ")
+ return errors.New("Failed to clear ALLOC ID resource pool")
+ }
+
+ if status := PONRMgr.ClearResourceIDPool(ctx, intfID, GEMPORT_ID); !status {
+ logger.Error(ctx, "Failed to clear GEMPORT ID resource pool")
+ return errors.New("Failed to clear GEMPORT ID resource pool")
+ }
+
+ if status := PONRMgr.ClearResourceIDPool(ctx, intfID, FLOW_ID); !status {
+ logger.Error(ctx, "Failed to clear FLOW ID resource pool")
+ return errors.New("Failed to clear FLOW ID resource pool")
+ }
+
+ 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
@@ -700,6 +757,13 @@
:return list/uint32/None: list, uint32 or None if resource type is
alloc_id/gemport_id, onu_id or invalid type respectively
*/
+
+ logger.Debugw(ctx, "getting-resource-id", log.Fields{
+ "intf-id": IntfID,
+ "resource-type": ResourceType,
+ "num": NumIDs,
+ })
+
if NumIDs < 1 {
logger.Error(ctx, "Invalid number of resources requested")
return nil, fmt.Errorf("Invalid number of resources requested %d", NumIDs)
@@ -768,21 +832,30 @@
return false
}
-func (PONRMgr *PONResourceManager) FreeResourceID(ctx context.Context, IntfID uint32, ResourceType string, ReleaseContent []uint32) bool {
+func (PONRMgr *PONResourceManager) FreeResourceID(ctx context.Context, IntfID uint32, ResourceType string, ReleaseContent []uint32) error {
/*
- 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
+ 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
*/
+
+ logger.Debugw(ctx, "freeing-resource-id", log.Fields{
+ "intf-id": IntfID,
+ "resource-type": ResourceType,
+ "release-content": ReleaseContent,
+ })
+
if !checkValidResourceType(ResourceType) {
- logger.Error(ctx, "Invalid resource type")
- return false
+ err := fmt.Errorf("Invalid resource type: %s", ResourceType)
+ logger.Error(ctx, err.Error())
+ return err
}
if ReleaseContent == nil {
- logger.Debug(ctx, "Nothing to release")
- return true
+ err := fmt.Errorf("Nothing to release")
+ logger.Debug(ctx, err.Error())
+ return err
}
// delegate to the master instance if sharing enabled across instances
SharedResourceMgr := PONRMgr.SharedResourceMgrs[PONRMgr.SharedIdxByType[ResourceType]]
@@ -791,22 +864,24 @@
}
Path := PONRMgr.GetPath(ctx, IntfID, ResourceType)
if Path == "" {
- logger.Error(ctx, "Failed to get path")
- return false
+ err := fmt.Errorf("Failed to get path for IntfId %d and ResourceType %s", IntfID, ResourceType)
+ logger.Error(ctx, err.Error())
+ return err
}
Resource, err := PONRMgr.GetResource(ctx, Path)
if err != nil {
- logger.Error(ctx, "Failed to get resource")
- return false
+ logger.Error(ctx, err.Error())
+ return err
}
for _, Val := range ReleaseContent {
PONRMgr.ReleaseID(ctx, Resource, Val)
}
if PONRMgr.UpdateResource(ctx, Path, Resource) != nil {
- logger.Errorf(ctx, "Free resource for %s failed", Path)
- return false
+ err := fmt.Errorf("Free resource for %s failed", Path)
+ logger.Errorf(ctx, err.Error())
+ return err
}
- return true
+ return nil
}
func (PONRMgr *PONResourceManager) UpdateResource(ctx context.Context, Path string, Resource map[string]interface{}) error {
@@ -1034,6 +1109,20 @@
return true
}
+func (PONRMgr *PONResourceManager) RemoveAllFlowIDInfo(ctx context.Context, IntfONUID string) bool {
+ /*
+ Remove flow_id_info details configured for the ONU.
+ :param pon_intf_onu_id: reference of PON interface id and onu id
+ */
+ Path := fmt.Sprintf(FLOW_ID_INFO_PATH_INTF_ONU_PREFIX, PONRMgr.DeviceID, IntfONUID)
+
+ if err := PONRMgr.KVStore.DeleteWithPrefix(ctx, Path); err != nil {
+ logger.Errorf(ctx, "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
@@ -1043,15 +1132,23 @@
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 {
- logger.Error(ctx, "failed to Marshal")
- return err
- }
+ if AllocIDs == nil {
+ // No more alloc ids associated with the key. Delete the key entirely
+ if err = PONRMgr.KVStore.Delete(ctx, Path); err != nil {
+ logger.Errorf(ctx, "Failed to delete key %s", Path)
+ return err
+ }
+ } else {
+ Value, err = json.Marshal(AllocIDs)
+ if err != nil {
+ logger.Error(ctx, "failed to Marshal")
+ return err
+ }
- if err = PONRMgr.KVStore.Put(ctx, Path, Value); err != nil {
- logger.Errorf(ctx, "Failed to update resource %s", Path)
- return err
+ if err = PONRMgr.KVStore.Put(ctx, Path, Value); err != nil {
+ logger.Errorf(ctx, "Failed to update resource %s", Path)
+ return err
+ }
}
return err
}
@@ -1067,15 +1164,23 @@
var err error
Path := fmt.Sprintf(GEMPORT_ID_RESOURCE_MAP_PATH, PONRMgr.DeviceID, IntfONUID)
logger.Debugf(ctx, "Updating gemport ids for %s", Path)
- Value, err = json.Marshal(GEMPortIDs)
- if err != nil {
- logger.Error(ctx, "failed to Marshal")
- return err
- }
+ if GEMPortIDs == nil {
+ // No more gemport ids associated with the key. Delete the key entirely
+ if err = PONRMgr.KVStore.Delete(ctx, Path); err != nil {
+ logger.Errorf(ctx, "Failed to delete key %s", Path)
+ return err
+ }
+ } else {
+ Value, err = json.Marshal(GEMPortIDs)
+ if err != nil {
+ logger.Error(ctx, "failed to Marshal")
+ return err
+ }
- if err = PONRMgr.KVStore.Put(ctx, Path, Value); err != nil {
- logger.Errorf(ctx, "Failed to update resource %s", Path)
- return err
+ if err = PONRMgr.KVStore.Put(ctx, Path, Value); err != nil {
+ logger.Errorf(ctx, "Failed to update resource %s", Path)
+ return err
+ }
}
return err
}
@@ -1168,12 +1273,10 @@
*/
ByteArray, err := ToByte(Resource[POOL])
if err != nil {
- logger.Error(ctx, "Failed to convert resource to byte array")
return 0, err
}
Data := bitmap.TSFromData(ByteArray, false)
if Data == nil {
- logger.Error(ctx, "Failed to get data from byte array")
return 0, errors.New("Failed to get data from byte array")
}
@@ -1184,6 +1287,9 @@
break
}
}
+ if Idx == Len {
+ return 0, errors.New("resource-exhausted--no-free-id-in-the-pool")
+ }
Data.Set(Idx, true)
res := uint32(Resource[START_IDX].(float64))
Resource[POOL] = Data.Data(false)
@@ -1208,6 +1314,10 @@
return false
}
Idx := Id - uint32(Resource[START_IDX].(float64))
+ if Idx >= uint32(Data.Len()) {
+ logger.Errorf(ctx, "ID %d is out of the boundaries of the pool", Id)
+ return false
+ }
Data.Set(int(Idx), false)
Resource[POOL] = Data.Data(false)
@@ -1225,6 +1335,10 @@
return false
}
Idx := Id - StartIndex
+ if Idx >= uint32(Data.Len()) {
+ logger.Errorf(ctx, "Reservation failed. ID %d is out of the boundaries of the pool", Id)
+ return false
+ }
Data.Set(int(Idx), true)
return true
}
@@ -1241,6 +1355,10 @@
return GEMPORT_ID
}
+func (PONRMgr *PONResourceManager) GetResourceTypeOnuID() string {
+ return ONU_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) {
@@ -1266,69 +1384,3 @@
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 {
- logger.Error(ctx, "failed to Marshal")
- return err
- }
-
- if err = PONRMgr.KVStore.Put(ctx, Path, Value); err != nil {
- logger.Errorf(ctx, "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 {
- logger.Errorw(ctx, "Failed to get from kv store", log.Fields{"path": path})
- return err
- } else if value == nil {
- logger.Debug(ctx, "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 {
- logger.Error(ctx, "Failed to convert to byte array")
- return err
- }
-
- if err = json.Unmarshal(Val, &onuGemInfo); err != nil {
- logger.Error(ctx, "Failed to unmarshall")
- return err
- }
- logger.Debugw(ctx, "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 {
- logger.Errorf(ctx, "Falied to remove resource %s", path)
- return err
- }
- return nil
-}
diff --git a/vendor/github.com/opencord/voltha-lib-go/v4/pkg/techprofile/4QueueHybridProfileMap1.json b/vendor/github.com/opencord/voltha-lib-go/v7/pkg/techprofile/4QueueHybridProfileMap1.json
similarity index 100%
rename from vendor/github.com/opencord/voltha-lib-go/v4/pkg/techprofile/4QueueHybridProfileMap1.json
rename to vendor/github.com/opencord/voltha-lib-go/v7/pkg/techprofile/4QueueHybridProfileMap1.json
diff --git a/vendor/github.com/opencord/voltha-lib-go/v4/pkg/techprofile/README.md b/vendor/github.com/opencord/voltha-lib-go/v7/pkg/techprofile/README.md
similarity index 100%
rename from vendor/github.com/opencord/voltha-lib-go/v4/pkg/techprofile/README.md
rename to vendor/github.com/opencord/voltha-lib-go/v7/pkg/techprofile/README.md
diff --git a/vendor/github.com/opencord/voltha-lib-go/v4/pkg/techprofile/SingleQueueEponProfile.json b/vendor/github.com/opencord/voltha-lib-go/v7/pkg/techprofile/SingleQueueEponProfile.json
similarity index 99%
rename from vendor/github.com/opencord/voltha-lib-go/v4/pkg/techprofile/SingleQueueEponProfile.json
rename to vendor/github.com/opencord/voltha-lib-go/v7/pkg/techprofile/SingleQueueEponProfile.json
index 00476a2..4015251 100644
--- a/vendor/github.com/opencord/voltha-lib-go/v4/pkg/techprofile/SingleQueueEponProfile.json
+++ b/vendor/github.com/opencord/voltha-lib-go/v7/pkg/techprofile/SingleQueueEponProfile.json
@@ -58,4 +58,4 @@
}
}
]
-}
\ No newline at end of file
+}
diff --git a/vendor/github.com/opencord/voltha-lib-go/v4/pkg/techprofile/common.go b/vendor/github.com/opencord/voltha-lib-go/v7/pkg/techprofile/common.go
similarity index 94%
rename from vendor/github.com/opencord/voltha-lib-go/v4/pkg/techprofile/common.go
rename to vendor/github.com/opencord/voltha-lib-go/v7/pkg/techprofile/common.go
index 544c780..df1b877 100644
--- a/vendor/github.com/opencord/voltha-lib-go/v4/pkg/techprofile/common.go
+++ b/vendor/github.com/opencord/voltha-lib-go/v7/pkg/techprofile/common.go
@@ -16,7 +16,7 @@
package techprofile
import (
- "github.com/opencord/voltha-lib-go/v4/pkg/log"
+ "github.com/opencord/voltha-lib-go/v7/pkg/log"
)
var logger log.CLogger
diff --git a/vendor/github.com/opencord/voltha-lib-go/v4/pkg/techprofile/config.go b/vendor/github.com/opencord/voltha-lib-go/v7/pkg/techprofile/config.go
similarity index 63%
rename from vendor/github.com/opencord/voltha-lib-go/v4/pkg/techprofile/config.go
rename to vendor/github.com/opencord/voltha-lib-go/v7/pkg/techprofile/config.go
index bcec75e..17980fa 100644
--- a/vendor/github.com/opencord/voltha-lib-go/v4/pkg/techprofile/config.go
+++ b/vendor/github.com/opencord/voltha-lib-go/v7/pkg/techprofile/config.go
@@ -16,8 +16,10 @@
package techprofile
import (
- "github.com/opencord/voltha-lib-go/v4/pkg/db"
+ "fmt"
"time"
+
+ "github.com/opencord/voltha-lib-go/v7/pkg/db"
)
// tech profile default constants
@@ -31,15 +33,19 @@
defaultKVStoreTimeout = 5 * time.Second //in seconds
- // Tech profile path prefix in kv store
- defaultKVPathPrefix = "service/voltha/technology_profiles"
+ // Tech profile path prefix in kv store (for the TP template)
+ // NOTE that this hardcoded to service/voltha as the TP template is shared across stacks
+ defaultTpKvPathPrefix = "service/voltha/technology_profiles"
+
+ // Tech profile path prefix in kv store (for TP instances)
+ defaultKVPathPrefix = "%s/technology_profiles"
+
+ // Resource instance path prefix in KV store (for Resource Instances)
+ defaultResourceInstancePathPrefix = "%s/resource_instances"
// 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
@@ -93,38 +99,42 @@
// TechprofileFlags represents the set of configurations used
type TechProfileFlags struct {
- KVStoreAddress string
- KVStoreType string
- KVStoreTimeout time.Duration
- KVBackend *db.Backend
- TPKVPathPrefix string
- TPFileKVPath string
- TPInstanceKVPath string
- DefaultTPName string
- TPVersion int
- NumGemPorts uint32
- DefaultPbits []string
- LogLevel int
- DefaultTechProfileID uint32
- DefaultNumGemPorts uint32
+ KVStoreAddress string
+ KVStoreType string
+ KVStoreTimeout time.Duration
+ KVBackend *db.Backend // this is the backend used to store TP instances
+ DefaultTpKVBackend *db.Backend // this is the backend used to read the TP profile
+ ResourceInstanceKVBacked *db.Backend // this is the backed used to read/write Resource Instances
+ TPKVPathPrefix string
+ defaultTpKvPathPrefix string
+ TPFileKVPath string
+ ResourceInstanceKVPathPrefix string
+ DefaultTPName string
+ TPVersion uint32
+ NumGemPorts uint32
+ DefaultPbits []string
+ LogLevel int
+ DefaultTechProfileID uint32
+ DefaultNumGemPorts uint32
}
-func NewTechProfileFlags(KVStoreType string, KVStoreAddress string) *TechProfileFlags {
+func NewTechProfileFlags(KVStoreType string, KVStoreAddress string, basePathKvStore string) *TechProfileFlags {
// initialize with default values
var techProfileFlags = TechProfileFlags{
- KVBackend: nil,
- KVStoreAddress: KVStoreAddress,
- 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,
+ KVBackend: nil,
+ KVStoreAddress: KVStoreAddress,
+ KVStoreType: KVStoreType,
+ KVStoreTimeout: defaultKVStoreTimeout,
+ DefaultTPName: defaultTechProfileName,
+ TPKVPathPrefix: fmt.Sprintf(defaultKVPathPrefix, basePathKvStore),
+ defaultTpKvPathPrefix: defaultTpKvPathPrefix,
+ TPVersion: defaultVersion,
+ TPFileKVPath: defaultTechProfileKVPath,
+ ResourceInstanceKVPathPrefix: fmt.Sprintf(defaultResourceInstancePathPrefix, basePathKvStore),
+ 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/v7/pkg/techprofile/tech_profile.go b/vendor/github.com/opencord/voltha-lib-go/v7/pkg/techprofile/tech_profile.go
new file mode 100644
index 0000000..32f73b9
--- /dev/null
+++ b/vendor/github.com/opencord/voltha-lib-go/v7/pkg/techprofile/tech_profile.go
@@ -0,0 +1,1498 @@
+/*
+ * 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 (
+ "bytes"
+ "context"
+ "errors"
+ "fmt"
+ "regexp"
+ "strconv"
+ "sync"
+ "time"
+
+ "github.com/golang/protobuf/jsonpb"
+ "github.com/golang/protobuf/proto"
+ "github.com/opencord/voltha-lib-go/v7/pkg/db"
+ "github.com/opencord/voltha-lib-go/v7/pkg/db/kvstore"
+ "github.com/opencord/voltha-lib-go/v7/pkg/log"
+ "github.com/opencord/voltha-protos/v5/go/openolt"
+ tp_pb "github.com/opencord/voltha-protos/v5/go/tech_profile"
+)
+
+// Interface to pon resource manager APIs
+type iPonResourceMgr interface {
+ GetResourceID(ctx context.Context, intfID uint32, resourceType string, numIDs uint32) ([]uint32, error)
+ FreeResourceID(ctx context.Context, intfID uint32, resourceType string, ReleaseContent []uint32) error
+ GetResourceTypeAllocID() string
+ GetResourceTypeGemPortID() string
+ GetResourceTypeOnuID() string
+ GetTechnology() string
+}
+
+type SchedulingPolicy int32
+
+const (
+ SchedulingPolicy_WRR SchedulingPolicy = 0
+ SchedulingPolicy_StrictPriority SchedulingPolicy = 1
+ SchedulingPolicy_Hybrid SchedulingPolicy = 2
+)
+
+type AdditionalBW int32
+
+const (
+ AdditionalBW_AdditionalBW_None AdditionalBW = 0
+ AdditionalBW_AdditionalBW_NA AdditionalBW = 1
+ AdditionalBW_AdditionalBW_BestEffort AdditionalBW = 2
+ AdditionalBW_AdditionalBW_Auto AdditionalBW = 3
+)
+
+type DiscardPolicy int32
+
+const (
+ DiscardPolicy_TailDrop DiscardPolicy = 0
+ DiscardPolicy_WTailDrop DiscardPolicy = 1
+ DiscardPolicy_Red DiscardPolicy = 2
+ DiscardPolicy_WRed DiscardPolicy = 3
+)
+
+// Required uniPortName format
+var uniPortNameFormatRegexp = regexp.MustCompile(`^olt-{[a-z0-9\-]+}/pon-{[0-9]+}/onu-{[0-9]+}/uni-{[0-9]+}$`)
+
+// instance control defaults
+const (
+ defaultOnuInstance = "multi-instance"
+ defaultUniInstance = "single-instance"
+ defaultGemPayloadSize = "auto"
+)
+
+// default discard config constants
+const (
+ defaultMinThreshold = 0
+ defaultMaxThreshold = 0
+ defaultMaxProbability = 0
+)
+
+// default scheduler contants
+const (
+ defaultPriority = 0
+ defaultWeight = 0
+)
+
+// default GEM attribute constants
+const (
+ defaultAESEncryption = "True"
+ defaultPriorityQueue = 0
+ defaultQueueWeight = 0
+ defaultMaxQueueSize = "auto"
+ defaultIsMulticast = "False"
+ defaultAccessControlList = "224.0.0.0-239.255.255.255"
+ defaultMcastGemID = 4069
+)
+
+// Default EPON constants
+const (
+ defaultPakageType = "B"
+)
+const (
+ defaultTrafficType = "BE"
+ defaultUnsolicitedGrantSize = 0
+ defaultNominalInterval = 0
+ defaultToleratedPollJitter = 0
+ defaultRequestTransmissionPolicy = 0
+ defaultNumQueueSet = 2
+)
+const (
+ defaultQThreshold1 = 5500
+ defaultQThreshold2 = 0
+ defaultQThreshold3 = 0
+ defaultQThreshold4 = 0
+ defaultQThreshold5 = 0
+ defaultQThreshold6 = 0
+ defaultQThreshold7 = 0
+)
+
+const (
+ xgspon = "XGS-PON"
+ xgpon = "XGPON"
+ gpon = "GPON"
+ epon = "EPON"
+)
+
+const (
+ MaxUniPortPerOnu = 16 // TODO: Adapter uses its own constant for MaxUniPort. How to synchronize this and have a single source of truth?
+)
+
+type TechProfileMgr struct {
+ config *TechProfileFlags
+ resourceMgr iPonResourceMgr
+ OnuIDMgmtLock sync.RWMutex
+ GemPortIDMgmtLock sync.RWMutex
+ AllocIDMgmtLock sync.RWMutex
+ tpInstanceMap map[string]*tp_pb.TechProfileInstance // Map of tp path to tp instance
+ tpInstanceMapLock sync.RWMutex
+ eponTpInstanceMap map[string]*tp_pb.EponTechProfileInstance // Map of tp path to epon tp instance
+ epontpInstanceMapLock sync.RWMutex
+ tpMap map[uint32]*tp_pb.TechProfile // Map of tp id to tp
+ tpMapLock sync.RWMutex
+ eponTpMap map[uint32]*tp_pb.EponTechProfile // map of tp id to epon tp
+ eponTpMapLock sync.RWMutex
+}
+
+func (t *TechProfileMgr) SetKVClient(ctx context.Context, pathPrefix string) *db.Backend {
+ kvClient, err := newKVClient(ctx, t.config.KVStoreType, t.config.KVStoreAddress, t.config.KVStoreTimeout)
+ if err != nil {
+ logger.Errorw(ctx, "failed-to-create-kv-client",
+ log.Fields{
+ "type": t.config.KVStoreType, "address": t.config.KVStoreAddress,
+ "timeout": t.config.KVStoreTimeout, "prefix": pathPrefix,
+ "error": err.Error(),
+ })
+ return nil
+ }
+ return &db.Backend{
+ Client: kvClient,
+ StoreType: t.config.KVStoreType,
+ Address: t.config.KVStoreAddress,
+ Timeout: t.config.KVStoreTimeout,
+ PathPrefix: pathPrefix}
+
+ /* 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 NewTechProfile(ctx context.Context, resourceMgr iPonResourceMgr, kvStoreType string, kvStoreAddress string, basePathKvStore string) (*TechProfileMgr, error) {
+ var techprofileObj TechProfileMgr
+ logger.Debug(ctx, "initializing-techprofile-mananger")
+ techprofileObj.config = NewTechProfileFlags(kvStoreType, kvStoreAddress, basePathKvStore)
+ techprofileObj.config.KVBackend = techprofileObj.SetKVClient(ctx, techprofileObj.config.TPKVPathPrefix)
+ techprofileObj.config.DefaultTpKVBackend = techprofileObj.SetKVClient(ctx, techprofileObj.config.defaultTpKvPathPrefix)
+ if techprofileObj.config.KVBackend == nil {
+ logger.Error(ctx, "failed-to-initialize-backend")
+ return nil, errors.New("kv-backend-init-failed")
+ }
+ techprofileObj.config.ResourceInstanceKVBacked = techprofileObj.SetKVClient(ctx, techprofileObj.config.ResourceInstanceKVPathPrefix)
+ if techprofileObj.config.ResourceInstanceKVBacked == nil {
+ logger.Error(ctx, "failed-to-initialize-resource-instance-kv-backend")
+ return nil, errors.New("resource-instance-kv-backend-init-failed")
+ }
+ techprofileObj.resourceMgr = resourceMgr
+ techprofileObj.tpInstanceMap = make(map[string]*tp_pb.TechProfileInstance)
+ techprofileObj.eponTpInstanceMap = make(map[string]*tp_pb.EponTechProfileInstance)
+ techprofileObj.tpMap = make(map[uint32]*tp_pb.TechProfile)
+ techprofileObj.eponTpMap = make(map[uint32]*tp_pb.EponTechProfile)
+ logger.Debug(ctx, "reconcile-tp-instance-cache-start")
+ if err := techprofileObj.reconcileTpInstancesToCache(ctx); err != nil {
+ logger.Errorw(ctx, "failed-to-reconcile-tp-instances", log.Fields{"err": err})
+ return nil, err
+ }
+ logger.Debug(ctx, "reconcile-tp-instance-cache-end")
+ logger.Debug(ctx, "initializing-tech-profile-manager-object-success")
+ return &techprofileObj, nil
+}
+
+// GetTechProfileInstanceKey returns the tp instance key that is used to reference TP Instance Map
+func (t *TechProfileMgr) GetTechProfileInstanceKey(ctx context.Context, tpID uint32, uniPortName string) string {
+ logger.Debugw(ctx, "get-tp-instance-kv-key", log.Fields{
+ "uniPortName": uniPortName,
+ "tpId": tpID,
+ })
+ // Make sure the uniPortName is as per format olt-{[a-z0-9\-]+}/pon-{[0-9]+}/onu-{[0-9]+}/uni-{[0-9]+}
+ if !uniPortNameFormatRegexp.Match([]byte(uniPortName)) {
+ logger.Warnw(ctx, "uni-port-name-not-confirming-to-format", log.Fields{"uniPortName": uniPortName})
+ }
+ // The key path prefix (like service/voltha/technology_profiles or service/voltha_voltha/technology_profiles)
+ // is expected to be attached by the components that use this path as part of the KVBackend configuration.
+ resourceInstanceKvPathSuffix := "%s/%d/%s" // <technology>/<tpID>/<uni-port-name>
+ // <uni-port-name> must be of the format pon-{\d+}/onu-{\d+}/uni-{\d+}
+ return fmt.Sprintf(resourceInstanceKvPathSuffix, t.resourceMgr.GetTechnology(), tpID, uniPortName)
+}
+
+// GetTPInstance gets TP instance from cache if found
+func (t *TechProfileMgr) GetTPInstance(ctx context.Context, path string) (interface{}, error) {
+ tech := t.resourceMgr.GetTechnology()
+ switch tech {
+ case xgspon, xgpon, gpon:
+ t.tpInstanceMapLock.RLock()
+ defer t.tpInstanceMapLock.RUnlock()
+ tpInst, ok := t.tpInstanceMap[path]
+ if !ok {
+ return nil, fmt.Errorf("tp-instance-not-found-tp-path-%v", path)
+ }
+ return tpInst, nil
+ case epon:
+ t.epontpInstanceMapLock.RLock()
+ defer t.epontpInstanceMapLock.RUnlock()
+ tpInst, ok := t.eponTpInstanceMap[path]
+ if !ok {
+ return nil, fmt.Errorf("tp-instance-not-found-tp-path-%v", path)
+ }
+ return tpInst, nil
+ default:
+ logger.Errorw(ctx, "unknown-tech", log.Fields{"tech": tech})
+ return nil, fmt.Errorf("unknown-tech-%s-tp-path-%v", tech, path)
+ }
+}
+
+// CreateTechProfileInstance creates a new TP instance.
+func (t *TechProfileMgr) CreateTechProfileInstance(ctx context.Context, tpID uint32, uniPortName string, intfID uint32) (interface{}, error) {
+ var tpInstance *tp_pb.TechProfileInstance
+ var eponTpInstance *tp_pb.EponTechProfileInstance
+
+ logger.Infow(ctx, "creating-tp-instance", log.Fields{"tpID": tpID, "uni": uniPortName, "intId": intfID})
+
+ // Make sure the uniPortName is as per format olt-{[a-z0-9\-]+}/pon-{[0-9]+}/onu-{[0-9]+}/uni-{[0-9]+}
+ if !uniPortNameFormatRegexp.Match([]byte(uniPortName)) {
+ logger.Errorw(ctx, "uni-port-name-not-confirming-to-format", log.Fields{"uniPortName": uniPortName})
+ return nil, fmt.Errorf("uni-port-name-not-confirming-to-format-%s", uniPortName)
+ }
+ tpInstancePathSuffix := t.GetTechProfileInstanceKey(ctx, tpID, uniPortName)
+
+ if t.resourceMgr.GetTechnology() == epon {
+ tp := t.getEponTPFromKVStore(ctx, tpID)
+ if tp != nil {
+ if err := t.validateInstanceControlAttr(ctx, *tp.InstanceControl); err != nil {
+ logger.Error(ctx, "invalid-instance-ctrl-attr-using-default-tp")
+ tp = t.getDefaultEponProfile(ctx)
+ } else {
+ logger.Infow(ctx, "using-specified-tp-from-kv-store", log.Fields{"tpID": tpID})
+ }
+ } else {
+ logger.Info(ctx, "tp-not-found-on-kv--creating-default-tp")
+ tp = t.getDefaultEponProfile(ctx)
+ }
+ // Store TP in cache
+ t.eponTpMapLock.Lock()
+ t.eponTpMap[tpID] = tp
+ t.eponTpMapLock.Unlock()
+
+ if eponTpInstance = t.allocateEponTPInstance(ctx, uniPortName, tp, intfID, tpInstancePathSuffix); eponTpInstance == nil {
+ logger.Error(ctx, "tp-instance-allocation-failed")
+ return nil, errors.New("tp-instance-allocation-failed")
+ }
+ t.epontpInstanceMapLock.Lock()
+ t.eponTpInstanceMap[tpInstancePathSuffix] = eponTpInstance
+ t.epontpInstanceMapLock.Unlock()
+ resInst := tp_pb.ResourceInstance{
+ TpId: tpID,
+ ProfileType: eponTpInstance.ProfileType,
+ SubscriberIdentifier: eponTpInstance.SubscriberIdentifier,
+ AllocId: eponTpInstance.AllocId,
+ }
+ for _, usQAttr := range eponTpInstance.UpstreamQueueAttributeList {
+ resInst.GemportIds = append(resInst.GemportIds, usQAttr.GemportId)
+ }
+
+ logger.Infow(ctx, "epon-tp-instance-created-successfully",
+ log.Fields{"tpID": tpID, "uni": uniPortName, "intfID": intfID})
+ if err := t.addResourceInstanceToKVStore(ctx, tpID, uniPortName, resInst); err != nil {
+ logger.Errorw(ctx, "failed-to-update-resource-instance-to-kv-store--freeing-up-resources", log.Fields{"err": err, "tpID": tpID, "uniPortName": uniPortName})
+ allocIDs := make([]uint32, 0)
+ allocIDs = append(allocIDs, resInst.AllocId)
+ errList := make([]error, 0)
+ errList = append(errList, t.FreeResourceID(ctx, intfID, t.resourceMgr.GetResourceTypeAllocID(), allocIDs))
+ errList = append(errList, t.FreeResourceID(ctx, intfID, t.resourceMgr.GetResourceTypeGemPortID(), resInst.GemportIds))
+ if len(errList) > 0 {
+ logger.Errorw(ctx, "failed-to-free-up-resources-on-kv-store--system-behavior-has-become-erratic", log.Fields{"tpID": tpID, "uniPortName": uniPortName, "errList": errList})
+ }
+ return nil, err
+ }
+ return eponTpInstance, nil
+ } else {
+ tp := t.getTPFromKVStore(ctx, tpID)
+ if tp != nil {
+ if err := t.validateInstanceControlAttr(ctx, *tp.InstanceControl); err != nil {
+ logger.Error(ctx, "invalid-instance-ctrl-attr--using-default-tp")
+ tp = t.getDefaultTechProfile(ctx)
+ } else {
+ logger.Infow(ctx, "using-specified-tp-from-kv-store", log.Fields{"tpID": tpID})
+ }
+ } else {
+ logger.Info(ctx, "tp-not-found-on-kv--creating-default-tp")
+ tp = t.getDefaultTechProfile(ctx)
+ }
+ // Store TP in cache
+ t.tpMapLock.Lock()
+ t.tpMap[tpID] = tp
+ t.tpMapLock.Unlock()
+
+ if tpInstance = t.allocateTPInstance(ctx, uniPortName, tp, intfID, tpInstancePathSuffix); tpInstance == nil {
+ logger.Error(ctx, "tp-instance-allocation-failed")
+ return nil, errors.New("tp-instance-allocation-failed")
+ }
+ t.tpInstanceMapLock.Lock()
+ t.tpInstanceMap[tpInstancePathSuffix] = tpInstance
+ t.tpInstanceMapLock.Unlock()
+
+ resInst := tp_pb.ResourceInstance{
+ TpId: tpID,
+ ProfileType: tpInstance.ProfileType,
+ SubscriberIdentifier: tpInstance.SubscriberIdentifier,
+ AllocId: tpInstance.UsScheduler.AllocId,
+ }
+ for _, usQAttr := range tpInstance.UpstreamGemPortAttributeList {
+ resInst.GemportIds = append(resInst.GemportIds, usQAttr.GemportId)
+ }
+
+ logger.Infow(ctx, "tp-instance-created-successfully",
+ log.Fields{"tpID": tpID, "uni": uniPortName, "intfID": intfID})
+ if err := t.addResourceInstanceToKVStore(ctx, tpID, uniPortName, resInst); err != nil {
+ logger.Errorw(ctx, "failed-to-update-resource-instance-to-kv-store--freeing-up-resources", log.Fields{"err": err, "tpID": tpID, "uniPortName": uniPortName})
+ allocIDs := make([]uint32, 0)
+ allocIDs = append(allocIDs, resInst.AllocId)
+ errList := make([]error, 0)
+ errList = append(errList, t.FreeResourceID(ctx, intfID, t.resourceMgr.GetResourceTypeAllocID(), allocIDs))
+ errList = append(errList, t.FreeResourceID(ctx, intfID, t.resourceMgr.GetResourceTypeGemPortID(), resInst.GemportIds))
+ if len(errList) > 0 {
+ logger.Fatalw(ctx, "failed-to-free-up-resources-on-kv-store--system-behavior-has-become-erratic", log.Fields{"err": err, "tpID": tpID, "uniPortName": uniPortName})
+ }
+ return nil, err
+ }
+
+ logger.Infow(ctx, "resource-instance-added-to-kv-store-successfully",
+ log.Fields{"tpID": tpID, "uni": uniPortName, "intfID": intfID})
+ return tpInstance, nil
+ }
+}
+
+// DeleteTechProfileInstance deletes the TP instance from the local cache as well as deletes the corresponding
+// resource instance from the KV store.
+func (t *TechProfileMgr) DeleteTechProfileInstance(ctx context.Context, tpID uint32, uniPortName string) error {
+ // Make sure the uniPortName is as per format olt-{[a-z0-9\-]+}/pon-{[0-9]+}/onu-{[0-9]+}/uni-{[0-9]+}
+ if !uniPortNameFormatRegexp.Match([]byte(uniPortName)) {
+ logger.Errorw(ctx, "uni-port-name-not-confirming-to-format", log.Fields{"uniPortName": uniPortName})
+ return fmt.Errorf("uni-port-name-not-confirming-to-format--%s", uniPortName)
+ }
+ path := t.GetTechProfileInstanceKey(ctx, tpID, uniPortName)
+ logger.Infow(ctx, "delete-tp-instance-from-cache", log.Fields{"key": path})
+ t.tpInstanceMapLock.Lock()
+ delete(t.tpInstanceMap, path)
+ t.tpInstanceMapLock.Unlock()
+ if err := t.removeResourceInstanceFromKVStore(ctx, tpID, uniPortName); err != nil {
+ return err
+ }
+ return nil
+}
+
+func (t *TechProfileMgr) GetMulticastTrafficQueues(ctx context.Context, tp *tp_pb.TechProfileInstance) []*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_DOWNSTREAM,
+ GemportId: tp.DownstreamGemPortAttributeList[Count].MulticastGemId,
+ PbitMap: tp.DownstreamGemPortAttributeList[Count].PbitMap,
+ AesEncryption: encryp,
+ SchedPolicy: tp.DownstreamGemPortAttributeList[Count].SchedulingPolicy,
+ Priority: tp.DownstreamGemPortAttributeList[Count].PriorityQ,
+ Weight: tp.DownstreamGemPortAttributeList[Count].Weight,
+ DiscardPolicy: tp.DownstreamGemPortAttributeList[Count].DiscardPolicy,
+ })
+ }
+ logger.Debugw(ctx, "Downstream Multicast Traffic queue list ", log.Fields{"queuelist": mcastTrafficQueues})
+ return mcastTrafficQueues
+}
+
+func (t *TechProfileMgr) GetGemportForPbit(ctx context.Context, tp interface{}, dir tp_pb.Direction, pbit uint32) interface{} {
+ /*
+ Function to get the Gemport mapped to a pbit.
+ */
+ switch tp := tp.(type) {
+ case *tp_pb.TechProfileInstance:
+ if dir == tp_pb.Direction_UPSTREAM {
+ // upstream GEM ports
+ numGemPorts := len(tp.UpstreamGemPortAttributeList)
+ for gemCnt := 0; gemCnt < numGemPorts; gemCnt++ {
+ lenOfPbitMap := len(tp.UpstreamGemPortAttributeList[gemCnt].PbitMap)
+ for pbitMapIdx := 2; pbitMapIdx < lenOfPbitMap; pbitMapIdx++ {
+ // Given a sample pbit map string "0b00000001", lenOfPbitMap is 10
+ // "lenOfPbitMap - pbitMapIdx + 1" will give pbit-i th value from LSB position in the pbit map string
+ if p, err := strconv.Atoi(string(tp.UpstreamGemPortAttributeList[gemCnt].PbitMap[lenOfPbitMap-pbitMapIdx+1])); err == nil {
+ if uint32(pbitMapIdx-2) == pbit && p == 1 { // Check this p-bit is set
+ logger.Debugw(ctx, "Found-US-GEMport-for-Pcp", log.Fields{"pbit": pbit, "GEMport": tp.UpstreamGemPortAttributeList[gemCnt].GemportId})
+ return tp.UpstreamGemPortAttributeList[gemCnt]
+ }
+ }
+ }
+ }
+ } else if dir == tp_pb.Direction_DOWNSTREAM {
+ //downstream GEM ports
+ numGemPorts := len(tp.DownstreamGemPortAttributeList)
+ for gemCnt := 0; gemCnt < numGemPorts; gemCnt++ {
+ lenOfPbitMap := len(tp.DownstreamGemPortAttributeList[gemCnt].PbitMap)
+ for pbitMapIdx := 2; pbitMapIdx < lenOfPbitMap; pbitMapIdx++ {
+ // Given a sample pbit map string "0b00000001", lenOfPbitMap is 10
+ // "lenOfPbitMap - pbitMapIdx + 1" will give pbit-i th value from LSB position in the pbit map string
+ if p, err := strconv.Atoi(string(tp.DownstreamGemPortAttributeList[gemCnt].PbitMap[lenOfPbitMap-pbitMapIdx+1])); err == nil {
+ if uint32(pbitMapIdx-2) == pbit && p == 1 { // Check this p-bit is set
+ logger.Debugw(ctx, "Found-DS-GEMport-for-Pcp", log.Fields{"pbit": pbit, "GEMport": tp.DownstreamGemPortAttributeList[gemCnt].GemportId})
+ return tp.DownstreamGemPortAttributeList[gemCnt]
+ }
+ }
+ }
+ }
+ }
+ logger.Errorw(ctx, "No-GemportId-Found-For-Pcp", log.Fields{"pcpVlan": pbit})
+ case *openolt.EponTechProfileInstance:
+ if dir == tp_pb.Direction_UPSTREAM {
+ // upstream GEM ports
+ numGemPorts := len(tp.UpstreamQueueAttributeList)
+ for gemCnt := 0; gemCnt < numGemPorts; gemCnt++ {
+ lenOfPbitMap := len(tp.UpstreamQueueAttributeList[gemCnt].PbitMap)
+ for pbitMapIdx := 2; pbitMapIdx < lenOfPbitMap; pbitMapIdx++ {
+ // Given a sample pbit map string "0b00000001", lenOfPbitMap is 10
+ // "lenOfPbitMap - pbitMapIdx + 1" will give pbit-i th value from LSB position in the pbit map string
+ if p, err := strconv.Atoi(string(tp.UpstreamQueueAttributeList[gemCnt].PbitMap[lenOfPbitMap-pbitMapIdx+1])); err == nil {
+ if uint32(pbitMapIdx-2) == pbit && p == 1 { // Check this p-bit is set
+ logger.Debugw(ctx, "Found-US-Queue-for-Pcp", log.Fields{"pbit": pbit, "Queue": tp.UpstreamQueueAttributeList[gemCnt].GemportId})
+ return tp.UpstreamQueueAttributeList[gemCnt]
+ }
+ }
+ }
+ }
+ } else if dir == tp_pb.Direction_DOWNSTREAM {
+ //downstream GEM ports
+ numGemPorts := len(tp.DownstreamQueueAttributeList)
+ for gemCnt := 0; gemCnt < numGemPorts; gemCnt++ {
+ lenOfPbitMap := len(tp.DownstreamQueueAttributeList[gemCnt].PbitMap)
+ for pbitMapIdx := 2; pbitMapIdx < lenOfPbitMap; pbitMapIdx++ {
+ // Given a sample pbit map string "0b00000001", lenOfPbitMap is 10
+ // "lenOfPbitMap - pbitMapIdx + 1" will give pbit-i th value from LSB position in the pbit map string
+ if p, err := strconv.Atoi(string(tp.DownstreamQueueAttributeList[gemCnt].PbitMap[lenOfPbitMap-pbitMapIdx+1])); err == nil {
+ if uint32(pbitMapIdx-2) == pbit && p == 1 { // Check this p-bit is set
+ logger.Debugw(ctx, "Found-DS-Queue-for-Pcp", log.Fields{"pbit": pbit, "Queue": tp.DownstreamQueueAttributeList[gemCnt].GemportId})
+ return tp.DownstreamQueueAttributeList[gemCnt]
+ }
+ }
+ }
+ }
+ }
+ logger.Errorw(ctx, "No-QueueId-Found-For-Pcp", log.Fields{"pcpVlan": pbit})
+ default:
+ logger.Errorw(ctx, "unknown-tech", log.Fields{"tp": tp})
+ }
+ return nil
+}
+
+// FindAllTpInstances returns all TechProfile instances for a given TechProfile table-id, pon interface ID and onu ID.
+func (t *TechProfileMgr) FindAllTpInstances(ctx context.Context, oltDeviceID string, tpID uint32, intfID uint32, onuID uint32) interface{} {
+ onuTpInstancePathSuffix := fmt.Sprintf("%s/%d/olt-{%s}/pon-{%d}/onu-{%d}", t.resourceMgr.GetTechnology(), tpID, oltDeviceID, intfID, onuID)
+ tech := t.resourceMgr.GetTechnology()
+ if tech == xgspon || tech == xgpon || tech == gpon {
+ t.tpInstanceMapLock.RLock()
+ defer t.tpInstanceMapLock.RUnlock()
+ tpInstancesTech := make([]tp_pb.TechProfileInstance, 0)
+ for i := 0; i < MaxUniPortPerOnu; i++ {
+ key := onuTpInstancePathSuffix + fmt.Sprintf("/uni-{%d}", i)
+ if tpInst, ok := t.tpInstanceMap[key]; ok {
+ tpInstancesTech = append(tpInstancesTech, *tpInst)
+ }
+ }
+ return tpInstancesTech
+ } else if tech == epon {
+ t.epontpInstanceMapLock.RLock()
+ defer t.epontpInstanceMapLock.RUnlock()
+ tpInstancesTech := make([]tp_pb.EponTechProfileInstance, 0)
+ for i := 0; i < MaxUniPortPerOnu; i++ {
+ key := onuTpInstancePathSuffix + fmt.Sprintf("/uni-{%d}", i)
+ if tpInst, ok := t.eponTpInstanceMap[key]; ok {
+ tpInstancesTech = append(tpInstancesTech, *tpInst)
+ }
+ }
+ return tpInstancesTech
+ } else {
+ logger.Errorw(ctx, "unknown-tech", log.Fields{"tech": tech, "tpID": tpID, "onuID": onuID, "intfID": intfID})
+ }
+ return nil
+}
+
+func (t *TechProfileMgr) GetResourceID(ctx context.Context, intfID uint32, resourceType string, numIDs uint32) ([]uint32, error) {
+ logger.Debugw(ctx, "getting-resource-id", log.Fields{
+ "intf-id": intfID,
+ "resource-type": resourceType,
+ "num": numIDs,
+ })
+ var err error
+ var ids []uint32
+ switch resourceType {
+ case t.resourceMgr.GetResourceTypeAllocID():
+ t.AllocIDMgmtLock.Lock()
+ ids, err = t.resourceMgr.GetResourceID(ctx, intfID, resourceType, numIDs)
+ t.AllocIDMgmtLock.Unlock()
+ case t.resourceMgr.GetResourceTypeGemPortID():
+ t.GemPortIDMgmtLock.Lock()
+ ids, err = t.resourceMgr.GetResourceID(ctx, intfID, resourceType, numIDs)
+ t.GemPortIDMgmtLock.Unlock()
+ case t.resourceMgr.GetResourceTypeOnuID():
+ t.OnuIDMgmtLock.Lock()
+ ids, err = t.resourceMgr.GetResourceID(ctx, intfID, resourceType, numIDs)
+ t.OnuIDMgmtLock.Unlock()
+ default:
+ return nil, fmt.Errorf("resourceType %s not supported", resourceType)
+ }
+ if err != nil {
+ return nil, err
+ }
+ return ids, nil
+}
+
+func (t *TechProfileMgr) FreeResourceID(ctx context.Context, intfID uint32, resourceType string, ReleaseContent []uint32) error {
+ logger.Debugw(ctx, "freeing-resource-id", log.Fields{
+ "intf-id": intfID,
+ "resource-type": resourceType,
+ "release-content": ReleaseContent,
+ })
+ var err error
+ switch resourceType {
+ case t.resourceMgr.GetResourceTypeAllocID():
+ t.AllocIDMgmtLock.Lock()
+ err = t.resourceMgr.FreeResourceID(ctx, intfID, resourceType, ReleaseContent)
+ t.AllocIDMgmtLock.Unlock()
+ case t.resourceMgr.GetResourceTypeGemPortID():
+ t.GemPortIDMgmtLock.Lock()
+ err = t.resourceMgr.FreeResourceID(ctx, intfID, resourceType, ReleaseContent)
+ t.GemPortIDMgmtLock.Unlock()
+ case t.resourceMgr.GetResourceTypeOnuID():
+ t.OnuIDMgmtLock.Lock()
+ err = t.resourceMgr.FreeResourceID(ctx, intfID, resourceType, ReleaseContent)
+ t.OnuIDMgmtLock.Unlock()
+ default:
+ return fmt.Errorf("resourceType %s not supported", resourceType)
+ }
+ if err != nil {
+ return err
+ }
+ return nil
+}
+
+func (t *TechProfileMgr) GetUsScheduler(tpInstance *tp_pb.TechProfileInstance) *tp_pb.SchedulerConfig {
+ return &tp_pb.SchedulerConfig{
+ Direction: tpInstance.UsScheduler.Direction,
+ AdditionalBw: tpInstance.UsScheduler.AdditionalBw,
+ Priority: tpInstance.UsScheduler.Priority,
+ Weight: tpInstance.UsScheduler.Weight,
+ SchedPolicy: tpInstance.UsScheduler.QSchedPolicy}
+}
+
+func (t *TechProfileMgr) GetDsScheduler(tpInstance *tp_pb.TechProfileInstance) *tp_pb.SchedulerConfig {
+ return &tp_pb.SchedulerConfig{
+ Direction: tpInstance.DsScheduler.Direction,
+ AdditionalBw: tpInstance.DsScheduler.AdditionalBw,
+ Priority: tpInstance.DsScheduler.Priority,
+ Weight: tpInstance.DsScheduler.Weight,
+ SchedPolicy: tpInstance.DsScheduler.QSchedPolicy}
+}
+
+func (t *TechProfileMgr) GetTrafficScheduler(tpInstance *tp_pb.TechProfileInstance, 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 (t *TechProfileMgr) GetTrafficQueues(ctx context.Context, tp *tp_pb.TechProfileInstance, direction tp_pb.Direction) ([]*tp_pb.TrafficQueue, error) {
+
+ var encryp bool
+ if direction == 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
+ }
+
+ GemPorts = append(GemPorts, &tp_pb.TrafficQueue{
+ Direction: direction,
+ GemportId: tp.UpstreamGemPortAttributeList[Count].GemportId,
+ PbitMap: tp.UpstreamGemPortAttributeList[Count].PbitMap,
+ AesEncryption: encryp,
+ SchedPolicy: tp.UpstreamGemPortAttributeList[Count].SchedulingPolicy,
+ Priority: tp.UpstreamGemPortAttributeList[Count].PriorityQ,
+ Weight: tp.UpstreamGemPortAttributeList[Count].Weight,
+ DiscardPolicy: tp.UpstreamGemPortAttributeList[Count].DiscardPolicy,
+ })
+ }
+ logger.Debugw(ctx, "Upstream Traffic queue list ", log.Fields{"queuelist": GemPorts})
+ return GemPorts, nil
+ } else if direction == 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
+ }
+
+ GemPorts = append(GemPorts, &tp_pb.TrafficQueue{
+ Direction: direction,
+ GemportId: tp.DownstreamGemPortAttributeList[Count].GemportId,
+ PbitMap: tp.DownstreamGemPortAttributeList[Count].PbitMap,
+ AesEncryption: encryp,
+ SchedPolicy: tp.DownstreamGemPortAttributeList[Count].SchedulingPolicy,
+ Priority: tp.DownstreamGemPortAttributeList[Count].PriorityQ,
+ Weight: tp.DownstreamGemPortAttributeList[Count].Weight,
+ DiscardPolicy: tp.DownstreamGemPortAttributeList[Count].DiscardPolicy,
+ })
+ }
+ logger.Debugw(ctx, "Downstream Traffic queue list ", log.Fields{"queuelist": GemPorts})
+ return GemPorts, nil
+ }
+
+ logger.Errorf(ctx, "Unsupported direction %s used for generating Traffic Queue list", direction)
+ return nil, fmt.Errorf("downstream gem port traffic queue creation failed due to unsupported direction %s", direction)
+}
+
+func (t *TechProfileMgr) validateInstanceControlAttr(ctx context.Context, instCtl tp_pb.InstanceControl) error {
+ if instCtl.Onu != "single-instance" && instCtl.Onu != "multi-instance" {
+ logger.Errorw(ctx, "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" {
+ logger.Errorw(ctx, "invalid-uni-instance-control-attribute", log.Fields{"uni-inst": instCtl.Uni})
+ return errors.New("invalid-uni-instance-ctl-attr")
+ }
+
+ if instCtl.Uni == "multi-instance" {
+ logger.Error(ctx, "uni-multi-instance-tp-not-supported")
+ return errors.New("uni-multi-instance-tp-not-supported")
+ }
+
+ return nil
+}
+
+// allocateTPInstance for GPON, XGPON and XGS-PON technology
+func (t *TechProfileMgr) allocateTPInstance(ctx context.Context, uniPortName string, tp *tp_pb.TechProfile, intfID uint32, tpInstPathSuffix string) *tp_pb.TechProfileInstance {
+
+ var usGemPortAttributeList []*tp_pb.GemPortAttributes
+ var dsGemPortAttributeList []*tp_pb.GemPortAttributes
+ var dsMulticastGemAttributeList []*tp_pb.GemPortAttributes
+ var dsUnicastGemAttributeList []*tp_pb.GemPortAttributes
+ var tcontIDs []uint32
+ var gemPorts []uint32
+ var err error
+
+ logger.Infow(ctx, "Allocating TechProfileMgr instance from techprofile template", log.Fields{"uniPortName": uniPortName, "intfID": intfID, "numGem": tp.NumGemPorts})
+
+ if tp.InstanceControl.Onu == "multi-instance" {
+ tcontIDs, err = t.GetResourceID(ctx, intfID, t.resourceMgr.GetResourceTypeAllocID(), 1)
+ if err != nil {
+ logger.Errorw(ctx, "Error getting alloc id from rsrcrMgr", log.Fields{"err": err, "intfID": intfID})
+ return nil
+ }
+ } else { // "single-instance"
+ if tpInst := t.getSingleInstanceTp(ctx, tpInstPathSuffix); tpInst == nil {
+ // No "single-instance" tp found on one any uni port for the given TP ID
+ // Allocate a new TcontID or AllocID
+ tcontIDs, err = t.GetResourceID(ctx, intfID, t.resourceMgr.GetResourceTypeAllocID(), 1)
+ if err != nil {
+ logger.Errorw(ctx, "Error getting alloc id from rsrcrMgr", log.Fields{"err": err, "intfID": intfID})
+ return nil
+ }
+ } else {
+ // Use the alloc-id from the existing TpInstance
+ tcontIDs = append(tcontIDs, tpInst.UsScheduler.AllocId)
+ }
+ }
+ logger.Debugw(ctx, "Num GEM ports in TP:", log.Fields{"NumGemPorts": tp.NumGemPorts})
+ gemPorts, err = t.GetResourceID(ctx, intfID, t.resourceMgr.GetResourceTypeGemPortID(), tp.NumGemPorts)
+ if err != nil {
+ logger.Errorw(ctx, "Error getting gemport ids from rsrcrMgr", log.Fields{"err": err, "intfID": intfID, "numGemports": tp.NumGemPorts})
+ return nil
+ }
+ logger.Infow(ctx, "Allocated tconts and GEM ports successfully", log.Fields{"tconts": tcontIDs, "gemports": gemPorts})
+ for index := 0; index < int(tp.NumGemPorts); index++ {
+ usGemPortAttributeList = append(usGemPortAttributeList,
+ &tp_pb.GemPortAttributes{GemportId: gemPorts[index],
+ MaxQSize: tp.UpstreamGemPortAttributeList[index].MaxQSize,
+ PbitMap: tp.UpstreamGemPortAttributeList[index].PbitMap,
+ AesEncryption: tp.UpstreamGemPortAttributeList[index].AesEncryption,
+ SchedulingPolicy: tp.UpstreamGemPortAttributeList[index].SchedulingPolicy,
+ PriorityQ: tp.UpstreamGemPortAttributeList[index].PriorityQ,
+ Weight: tp.UpstreamGemPortAttributeList[index].Weight,
+ DiscardPolicy: tp.UpstreamGemPortAttributeList[index].DiscardPolicy,
+ DiscardConfig: tp.UpstreamGemPortAttributeList[index].DiscardConfig})
+ }
+
+ logger.Info(ctx, "length of DownstreamGemPortAttributeList", len(tp.DownstreamGemPortAttributeList))
+ //put multicast and unicast downstream GEM port attributes in different lists first
+ for index := 0; index < len(tp.DownstreamGemPortAttributeList); index++ {
+ if isMulticastGem(tp.DownstreamGemPortAttributeList[index].IsMulticast) {
+ dsMulticastGemAttributeList = append(dsMulticastGemAttributeList,
+ &tp_pb.GemPortAttributes{
+ MulticastGemId: tp.DownstreamGemPortAttributeList[index].MulticastGemId,
+ MaxQSize: tp.DownstreamGemPortAttributeList[index].MaxQSize,
+ PbitMap: tp.DownstreamGemPortAttributeList[index].PbitMap,
+ AesEncryption: tp.DownstreamGemPortAttributeList[index].AesEncryption,
+ SchedulingPolicy: tp.DownstreamGemPortAttributeList[index].SchedulingPolicy,
+ PriorityQ: tp.DownstreamGemPortAttributeList[index].PriorityQ,
+ Weight: tp.DownstreamGemPortAttributeList[index].Weight,
+ DiscardPolicy: tp.DownstreamGemPortAttributeList[index].DiscardPolicy,
+ DiscardConfig: tp.DownstreamGemPortAttributeList[index].DiscardConfig,
+ IsMulticast: tp.DownstreamGemPortAttributeList[index].IsMulticast,
+ DynamicAccessControlList: tp.DownstreamGemPortAttributeList[index].DynamicAccessControlList,
+ StaticAccessControlList: tp.DownstreamGemPortAttributeList[index].StaticAccessControlList})
+ } else {
+ dsUnicastGemAttributeList = append(dsUnicastGemAttributeList,
+ &tp_pb.GemPortAttributes{
+ MaxQSize: tp.DownstreamGemPortAttributeList[index].MaxQSize,
+ PbitMap: tp.DownstreamGemPortAttributeList[index].PbitMap,
+ AesEncryption: tp.DownstreamGemPortAttributeList[index].AesEncryption,
+ SchedulingPolicy: tp.DownstreamGemPortAttributeList[index].SchedulingPolicy,
+ PriorityQ: tp.DownstreamGemPortAttributeList[index].PriorityQ,
+ Weight: tp.DownstreamGemPortAttributeList[index].Weight,
+ DiscardPolicy: tp.DownstreamGemPortAttributeList[index].DiscardPolicy,
+ DiscardConfig: tp.DownstreamGemPortAttributeList[index].DiscardConfig})
+ }
+ }
+ //add unicast downstream GEM ports to dsGemPortAttributeList
+ if dsUnicastGemAttributeList != nil {
+ for index := 0; index < int(tp.NumGemPorts); index++ {
+ dsGemPortAttributeList = append(dsGemPortAttributeList,
+ &tp_pb.GemPortAttributes{GemportId: gemPorts[index],
+ MaxQSize: dsUnicastGemAttributeList[index].MaxQSize,
+ PbitMap: dsUnicastGemAttributeList[index].PbitMap,
+ AesEncryption: dsUnicastGemAttributeList[index].AesEncryption,
+ SchedulingPolicy: dsUnicastGemAttributeList[index].SchedulingPolicy,
+ PriorityQ: dsUnicastGemAttributeList[index].PriorityQ,
+ 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 &tp_pb.TechProfileInstance{
+ SubscriberIdentifier: uniPortName,
+ Name: tp.Name,
+ ProfileType: tp.ProfileType,
+ Version: tp.Version,
+ NumGemPorts: tp.NumGemPorts,
+ InstanceControl: tp.InstanceControl,
+ UsScheduler: &tp_pb.SchedulerAttributes{
+ AllocId: tcontIDs[0],
+ Direction: tp.UsScheduler.Direction,
+ AdditionalBw: tp.UsScheduler.AdditionalBw,
+ Priority: tp.UsScheduler.Priority,
+ Weight: tp.UsScheduler.Weight,
+ QSchedPolicy: tp.UsScheduler.QSchedPolicy},
+ DsScheduler: &tp_pb.SchedulerAttributes{
+ 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}
+}
+
+// allocateTPInstance function for EPON
+func (t *TechProfileMgr) allocateEponTPInstance(ctx context.Context, uniPortName string, tp *tp_pb.EponTechProfile, intfID uint32, tpInstPath string) *tp_pb.EponTechProfileInstance {
+
+ var usQueueAttributeList []*tp_pb.EPONQueueAttributes
+ var dsQueueAttributeList []*tp_pb.EPONQueueAttributes
+ var tcontIDs []uint32
+ var gemPorts []uint32
+ var err error
+
+ logger.Infow(ctx, "allocating-tp-instance-from-tp-template", log.Fields{"uniPortName": uniPortName, "intfID": intfID, "numGem": tp.NumGemPorts})
+
+ if tp.InstanceControl.Onu == "multi-instance" {
+ if tcontIDs, err = t.GetResourceID(ctx, intfID, t.resourceMgr.GetResourceTypeAllocID(), 1); err != nil {
+ logger.Errorw(ctx, "Error getting alloc id from rsrcrMgr", log.Fields{"err": err, "intfID": intfID})
+ return nil
+ }
+ } else { // "single-instance"
+ if tpInst := t.getSingleInstanceEponTp(ctx, tpInstPath); 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.GetResourceID(ctx, intfID, t.resourceMgr.GetResourceTypeAllocID(), 1); err != nil {
+ logger.Errorw(ctx, "error-getting-alloc-id-from-resource-mgr", log.Fields{"err": err, "intfID": intfID})
+ return nil
+ }
+ } else {
+ // Use the alloc-id from the existing TpInstance
+ tcontIDs = append(tcontIDs, tpInst.AllocId)
+ }
+ }
+ logger.Debugw(ctx, "Num GEM ports in TP:", log.Fields{"NumGemPorts": tp.NumGemPorts})
+ if gemPorts, err = t.GetResourceID(ctx, intfID, t.resourceMgr.GetResourceTypeGemPortID(), tp.NumGemPorts); err != nil {
+ logger.Errorw(ctx, "error-getting-gemport-id-from-resource-mgr", log.Fields{"err": err, "intfID": intfID, "numGemports": tp.NumGemPorts})
+ return nil
+ }
+ logger.Infow(ctx, "allocated-alloc-id-and-gemport-successfully", log.Fields{"tconts": tcontIDs, "gemports": gemPorts})
+ for index := 0; index < int(tp.NumGemPorts); index++ {
+ usQueueAttributeList = append(usQueueAttributeList,
+ &tp_pb.EPONQueueAttributes{GemportId: gemPorts[index],
+ MaxQSize: tp.UpstreamQueueAttributeList[index].MaxQSize,
+ PbitMap: tp.UpstreamQueueAttributeList[index].PbitMap,
+ AesEncryption: tp.UpstreamQueueAttributeList[index].AesEncryption,
+ TrafficType: tp.UpstreamQueueAttributeList[index].TrafficType,
+ UnsolicitedGrantSize: tp.UpstreamQueueAttributeList[index].UnsolicitedGrantSize,
+ NominalInterval: tp.UpstreamQueueAttributeList[index].NominalInterval,
+ ToleratedPollJitter: tp.UpstreamQueueAttributeList[index].ToleratedPollJitter,
+ RequestTransmissionPolicy: tp.UpstreamQueueAttributeList[index].RequestTransmissionPolicy,
+ NumQSets: tp.UpstreamQueueAttributeList[index].NumQSets,
+ QThresholds: tp.UpstreamQueueAttributeList[index].QThresholds,
+ SchedulingPolicy: tp.UpstreamQueueAttributeList[index].SchedulingPolicy,
+ PriorityQ: tp.UpstreamQueueAttributeList[index].PriorityQ,
+ Weight: tp.UpstreamQueueAttributeList[index].Weight,
+ DiscardPolicy: tp.UpstreamQueueAttributeList[index].DiscardPolicy,
+ DiscardConfig: tp.UpstreamQueueAttributeList[index].DiscardConfig})
+ }
+
+ logger.Info(ctx, "length-of-downstream-gemport-attribute-list", len(tp.DownstreamQueueAttributeList))
+ for index := 0; index < int(tp.NumGemPorts); index++ {
+ dsQueueAttributeList = append(dsQueueAttributeList,
+ &tp_pb.EPONQueueAttributes{GemportId: gemPorts[index],
+ MaxQSize: tp.DownstreamQueueAttributeList[index].MaxQSize,
+ PbitMap: tp.DownstreamQueueAttributeList[index].PbitMap,
+ AesEncryption: tp.DownstreamQueueAttributeList[index].AesEncryption,
+ SchedulingPolicy: tp.DownstreamQueueAttributeList[index].SchedulingPolicy,
+ PriorityQ: tp.DownstreamQueueAttributeList[index].PriorityQ,
+ Weight: tp.DownstreamQueueAttributeList[index].Weight,
+ DiscardPolicy: tp.DownstreamQueueAttributeList[index].DiscardPolicy,
+ DiscardConfig: tp.DownstreamQueueAttributeList[index].DiscardConfig})
+ }
+
+ return &tp_pb.EponTechProfileInstance{
+ SubscriberIdentifier: uniPortName,
+ Name: tp.Name,
+ ProfileType: tp.ProfileType,
+ Version: tp.Version,
+ NumGemPorts: tp.NumGemPorts,
+ InstanceControl: tp.InstanceControl,
+ PackageType: tp.PackageType,
+ AllocId: tcontIDs[0],
+ UpstreamQueueAttributeList: usQueueAttributeList,
+ DownstreamQueueAttributeList: dsQueueAttributeList}
+}
+
+// getSingleInstanceTp returns another TpInstance (GPON, XGPON, XGS-PON) 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, tpPathSuffix string) *tp_pb.TechProfileInstance {
+
+ // For example:
+ // tpPathSuffix like "XGS-PON/64/olt-{1234}/pon-{0}/onu-{1}/uni-{1}"
+ // is broken into ["XGS-PON/64/olt-{1234}/pon-{0}/onu-{1}" ""]
+ uniPathSlice := regexp.MustCompile(`/uni-{[0-9]+}$`).Split(tpPathSuffix, 2)
+
+ t.tpInstanceMapLock.RLock()
+ defer t.tpInstanceMapLock.RUnlock()
+ for i := 0; i < MaxUniPortPerOnu; i++ {
+ key := fmt.Sprintf(uniPathSlice[0]+"/uni-{%d}", i)
+ if tpInst, ok := t.tpInstanceMap[key]; ok {
+ logger.Debugw(ctx, "found-single-instance-tp", log.Fields{"key": key})
+ return tpInst
+ }
+ }
+ return nil
+}
+
+// getSingleInstanceTp returns another TpInstance (EPON) for an ONU on a different
+// uni port for the same TP ID, if it finds one, else nil.
+func (t *TechProfileMgr) getSingleInstanceEponTp(ctx context.Context, tpPathSuffix string) *tp_pb.EponTechProfileInstance {
+ // For example:
+ // tpPathSuffix like "EPON/64/olt-{1234}/pon-{0}/onu-{1}/uni-{1}"
+ // is broken into ["EPON/64/-{1234}/pon-{0}/onu-{1}" ""]
+ uniPathSlice := regexp.MustCompile(`/uni-{[0-9]+}$`).Split(tpPathSuffix, 2)
+
+ t.epontpInstanceMapLock.RLock()
+ defer t.epontpInstanceMapLock.RUnlock()
+ for i := 0; i < MaxUniPortPerOnu; i++ {
+ key := fmt.Sprintf(uniPathSlice[0]+"/uni-{%d}", i)
+ if tpInst, ok := t.eponTpInstanceMap[key]; ok {
+ logger.Debugw(ctx, "found-single-instance-tp", log.Fields{"key": key})
+ return tpInst
+ }
+ }
+ return nil
+}
+
+// getDefaultTechProfile returns a default TechProfile for GPON, XGPON, XGS-PON
+func (t *TechProfileMgr) getDefaultTechProfile(ctx context.Context) *tp_pb.TechProfile {
+ var usGemPortAttributeList []*tp_pb.GemPortAttributes
+ var dsGemPortAttributeList []*tp_pb.GemPortAttributes
+
+ for _, pbit := range t.config.DefaultPbits {
+ logger.Debugw(ctx, "creating-gem-port-profile-profile", log.Fields{"pbit": pbit})
+ usGemPortAttributeList = append(usGemPortAttributeList,
+ &tp_pb.GemPortAttributes{
+ MaxQSize: defaultMaxQueueSize,
+ PbitMap: pbit,
+ AesEncryption: defaultAESEncryption,
+ SchedulingPolicy: tp_pb.SchedulingPolicy_WRR,
+ PriorityQ: defaultPriorityQueue,
+ Weight: defaultQueueWeight,
+ DiscardPolicy: tp_pb.DiscardPolicy_TailDrop,
+ DiscardConfigV2: &tp_pb.DiscardConfig{
+ DiscardPolicy: tp_pb.DiscardPolicy_Red,
+ DiscardConfig: &tp_pb.DiscardConfig_RedDiscardConfig{
+ RedDiscardConfig: &tp_pb.RedDiscardConfig{
+ MinThreshold: defaultMinThreshold,
+ MaxThreshold: defaultMaxThreshold,
+ MaxProbability: defaultMaxProbability,
+ },
+ },
+ },
+ DiscardConfig: &tp_pb.RedDiscardConfig{
+ MinThreshold: defaultMinThreshold,
+ MaxThreshold: defaultMaxThreshold,
+ MaxProbability: defaultMaxProbability,
+ },
+ })
+ dsGemPortAttributeList = append(dsGemPortAttributeList,
+ &tp_pb.GemPortAttributes{
+ MaxQSize: defaultMaxQueueSize,
+ PbitMap: pbit,
+ AesEncryption: defaultAESEncryption,
+ SchedulingPolicy: tp_pb.SchedulingPolicy_WRR,
+ PriorityQ: defaultPriorityQueue,
+ Weight: defaultQueueWeight,
+ DiscardPolicy: tp_pb.DiscardPolicy_TailDrop,
+ DiscardConfigV2: &tp_pb.DiscardConfig{
+ DiscardPolicy: tp_pb.DiscardPolicy_Red,
+ DiscardConfig: &tp_pb.DiscardConfig_RedDiscardConfig{
+ RedDiscardConfig: &tp_pb.RedDiscardConfig{
+ MinThreshold: defaultMinThreshold,
+ MaxThreshold: defaultMaxThreshold,
+ MaxProbability: defaultMaxProbability,
+ },
+ },
+ },
+ DiscardConfig: &tp_pb.RedDiscardConfig{
+ MinThreshold: defaultMinThreshold,
+ MaxThreshold: defaultMaxThreshold,
+ MaxProbability: defaultMaxProbability,
+ },
+ IsMulticast: defaultIsMulticast,
+ DynamicAccessControlList: defaultAccessControlList,
+ StaticAccessControlList: defaultAccessControlList,
+ MulticastGemId: defaultMcastGemID})
+ }
+ return &tp_pb.TechProfile{
+ Name: t.config.DefaultTPName,
+ ProfileType: t.resourceMgr.GetTechnology(),
+ Version: t.config.TPVersion,
+ NumGemPorts: uint32(len(usGemPortAttributeList)),
+ InstanceControl: &tp_pb.InstanceControl{
+ Onu: defaultOnuInstance,
+ Uni: defaultUniInstance,
+ MaxGemPayloadSize: defaultGemPayloadSize},
+ UsScheduler: &tp_pb.SchedulerAttributes{
+ Direction: tp_pb.Direction_UPSTREAM,
+ AdditionalBw: tp_pb.AdditionalBW_AdditionalBW_BestEffort,
+ Priority: defaultPriority,
+ Weight: defaultWeight,
+ QSchedPolicy: tp_pb.SchedulingPolicy_Hybrid},
+ DsScheduler: &tp_pb.SchedulerAttributes{
+ Direction: tp_pb.Direction_DOWNSTREAM,
+ AdditionalBw: tp_pb.AdditionalBW_AdditionalBW_BestEffort,
+ Priority: defaultPriority,
+ Weight: defaultWeight,
+ QSchedPolicy: tp_pb.SchedulingPolicy_Hybrid},
+ UpstreamGemPortAttributeList: usGemPortAttributeList,
+ DownstreamGemPortAttributeList: dsGemPortAttributeList}
+}
+
+// getDefaultEponProfile returns a default TechProfile for EPON
+func (t *TechProfileMgr) getDefaultEponProfile(ctx context.Context) *tp_pb.EponTechProfile {
+
+ var usQueueAttributeList []*tp_pb.EPONQueueAttributes
+ var dsQueueAttributeList []*tp_pb.EPONQueueAttributes
+
+ for _, pbit := range t.config.DefaultPbits {
+ logger.Debugw(ctx, "Creating Queue", log.Fields{"pbit": pbit})
+ usQueueAttributeList = append(usQueueAttributeList,
+ &tp_pb.EPONQueueAttributes{
+ MaxQSize: defaultMaxQueueSize,
+ PbitMap: pbit,
+ AesEncryption: defaultAESEncryption,
+ TrafficType: defaultTrafficType,
+ UnsolicitedGrantSize: defaultUnsolicitedGrantSize,
+ NominalInterval: defaultNominalInterval,
+ ToleratedPollJitter: defaultToleratedPollJitter,
+ RequestTransmissionPolicy: defaultRequestTransmissionPolicy,
+ NumQSets: defaultNumQueueSet,
+ QThresholds: &tp_pb.QThresholds{
+ QThreshold1: defaultQThreshold1,
+ QThreshold2: defaultQThreshold2,
+ QThreshold3: defaultQThreshold3,
+ QThreshold4: defaultQThreshold4,
+ QThreshold5: defaultQThreshold5,
+ QThreshold6: defaultQThreshold6,
+ QThreshold7: defaultQThreshold7},
+ SchedulingPolicy: tp_pb.SchedulingPolicy_WRR,
+ PriorityQ: defaultPriorityQueue,
+ Weight: defaultQueueWeight,
+ DiscardPolicy: tp_pb.DiscardPolicy_TailDrop,
+ DiscardConfigV2: &tp_pb.DiscardConfig{
+ DiscardPolicy: tp_pb.DiscardPolicy_Red,
+ DiscardConfig: &tp_pb.DiscardConfig_RedDiscardConfig{
+ RedDiscardConfig: &tp_pb.RedDiscardConfig{
+ MinThreshold: defaultMinThreshold,
+ MaxThreshold: defaultMaxThreshold,
+ MaxProbability: defaultMaxProbability,
+ },
+ },
+ },
+ DiscardConfig: &tp_pb.RedDiscardConfig{
+ MinThreshold: defaultMinThreshold,
+ MaxThreshold: defaultMaxThreshold,
+ MaxProbability: defaultMaxProbability,
+ }})
+ dsQueueAttributeList = append(dsQueueAttributeList,
+ &tp_pb.EPONQueueAttributes{
+ MaxQSize: defaultMaxQueueSize,
+ PbitMap: pbit,
+ AesEncryption: defaultAESEncryption,
+ SchedulingPolicy: tp_pb.SchedulingPolicy_WRR,
+ PriorityQ: defaultPriorityQueue,
+ Weight: defaultQueueWeight,
+ DiscardPolicy: tp_pb.DiscardPolicy_TailDrop,
+ DiscardConfigV2: &tp_pb.DiscardConfig{
+ DiscardPolicy: tp_pb.DiscardPolicy_Red,
+ DiscardConfig: &tp_pb.DiscardConfig_RedDiscardConfig{
+ RedDiscardConfig: &tp_pb.RedDiscardConfig{
+ MinThreshold: defaultMinThreshold,
+ MaxThreshold: defaultMaxThreshold,
+ MaxProbability: defaultMaxProbability,
+ },
+ },
+ },
+ DiscardConfig: &tp_pb.RedDiscardConfig{
+ MinThreshold: defaultMinThreshold,
+ MaxThreshold: defaultMaxThreshold,
+ MaxProbability: defaultMaxProbability,
+ }})
+ }
+ return &tp_pb.EponTechProfile{
+ Name: t.config.DefaultTPName,
+ ProfileType: t.resourceMgr.GetTechnology(),
+ Version: t.config.TPVersion,
+ NumGemPorts: uint32(len(usQueueAttributeList)),
+ InstanceControl: &tp_pb.InstanceControl{
+ Onu: defaultOnuInstance,
+ Uni: defaultUniInstance,
+ MaxGemPayloadSize: defaultGemPayloadSize},
+ PackageType: defaultPakageType,
+ UpstreamQueueAttributeList: usQueueAttributeList,
+ DownstreamQueueAttributeList: dsQueueAttributeList}
+}
+
+//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 (t *TechProfileMgr) addResourceInstanceToKVStore(ctx context.Context, tpID uint32, uniPortName string, resInst tp_pb.ResourceInstance) error {
+ logger.Debugw(ctx, "adding-resource-instance-to-kv-store", log.Fields{"tpID": tpID, "uniPortName": uniPortName, "resInst": resInst})
+ val, err := proto.Marshal(&resInst)
+ if err != nil {
+ logger.Errorw(ctx, "failed-to-marshall-resource-instance", log.Fields{"err": err, "tpID": tpID, "uniPortName": uniPortName, "resInst": resInst})
+ return err
+ }
+ err = t.config.ResourceInstanceKVBacked.Put(ctx, fmt.Sprintf("%s/%d/%s", t.resourceMgr.GetTechnology(), tpID, uniPortName), val)
+ return err
+}
+
+func (t *TechProfileMgr) removeResourceInstanceFromKVStore(ctx context.Context, tpID uint32, uniPortName string) error {
+ logger.Debugw(ctx, "removing-resource-instance-to-kv-store", log.Fields{"tpID": tpID, "uniPortName": uniPortName})
+ if err := t.config.ResourceInstanceKVBacked.Delete(ctx, fmt.Sprintf("%s/%d/%s", t.resourceMgr.GetTechnology(), tpID, uniPortName)); err != nil {
+ logger.Errorw(ctx, "error-removing-resource-instance-to-kv-store", log.Fields{"err": err, "tpID": tpID, "uniPortName": uniPortName})
+ return err
+ }
+ return nil
+}
+
+func (t *TechProfileMgr) getTPFromKVStore(ctx context.Context, tpID uint32) *tp_pb.TechProfile {
+ var tp *tp_pb.TechProfile
+ t.tpMapLock.RLock()
+ tp, ok := t.tpMap[tpID]
+ t.tpMapLock.RUnlock()
+ if ok {
+ logger.Debugw(ctx, "found-tp-in-cache", log.Fields{"tpID": tpID})
+ return tp
+ }
+ key := fmt.Sprintf(t.config.TPFileKVPath, t.resourceMgr.GetTechnology(), tpID)
+ logger.Debugw(ctx, "getting-tp-from-kv-store", log.Fields{"tpID": tpID, "Key": key})
+ kvresult, err := t.config.DefaultTpKVBackend.Get(ctx, key)
+ if err != nil {
+ logger.Errorw(ctx, "error-fetching-from-kv-store", log.Fields{"err": err, "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 {
+ lTp := &tp_pb.TechProfile{}
+ reader := bytes.NewReader(value)
+ if err = jsonpb.Unmarshal(reader, lTp); err != nil {
+ logger.Errorw(ctx, "error-unmarshalling-tp-from-kv-store", log.Fields{"err": err, "tpID": tpID, "error": err})
+ return nil
+ }
+
+ logger.Debugw(ctx, "success-fetched-tp-from-kv-store", log.Fields{"tpID": tpID, "value": *lTp})
+ return lTp
+ } else {
+ logger.Errorw(ctx, "error-decoding-tp", log.Fields{"err": err, "tpID": tpID})
+ // We we create a default profile in this case.
+ }
+ }
+
+ return nil
+}
+
+func (t *TechProfileMgr) getEponTPFromKVStore(ctx context.Context, tpID uint32) *tp_pb.EponTechProfile {
+ var eponTp *tp_pb.EponTechProfile
+ t.eponTpMapLock.RLock()
+ eponTp, ok := t.eponTpMap[tpID]
+ t.eponTpMapLock.RUnlock()
+ if ok {
+ logger.Debugw(ctx, "found-tp-in-cache", log.Fields{"tpID": tpID})
+ return eponTp
+ }
+ key := fmt.Sprintf(t.config.TPFileKVPath, t.resourceMgr.GetTechnology(), tpID)
+ logger.Debugw(ctx, "getting-epon-tp-from-kv-store", log.Fields{"tpID": tpID, "Key": key})
+ kvresult, err := t.config.DefaultTpKVBackend.Get(ctx, key)
+ if err != nil {
+ logger.Errorw(ctx, "error-fetching-from-kv-store", log.Fields{"err": err, "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 {
+ lEponTp := &tp_pb.EponTechProfile{}
+ reader := bytes.NewReader(value)
+ if err = jsonpb.Unmarshal(reader, lEponTp); err != nil {
+ logger.Errorw(ctx, "error-unmarshalling-epon-tp-from-kv-store", log.Fields{"err": err, "tpID": tpID, "error": err})
+ return nil
+ }
+
+ logger.Debugw(ctx, "success-fetching-epon-tp-from-kv-store", log.Fields{"tpID": tpID, "value": *lEponTp})
+ return lEponTp
+ }
+ }
+ return nil
+}
+
+func newKVClient(ctx context.Context, storeType string, address string, timeout time.Duration) (kvstore.Client, error) {
+
+ logger.Infow(ctx, "kv-store", log.Fields{"storeType": storeType, "address": address})
+ switch storeType {
+ case "etcd":
+ return kvstore.NewEtcdClient(ctx, address, timeout, log.WarnLevel)
+ }
+ return nil, errors.New("unsupported-kv-store")
+}
+
+// buildTpInstanceFromResourceInstance for GPON, XGPON and XGS-PON technology - build TpInstance from TechProfile template and ResourceInstance
+func (t *TechProfileMgr) buildTpInstanceFromResourceInstance(ctx context.Context, tp *tp_pb.TechProfile, resInst *tp_pb.ResourceInstance) *tp_pb.TechProfileInstance {
+
+ var usGemPortAttributeList []*tp_pb.GemPortAttributes
+ var dsGemPortAttributeList []*tp_pb.GemPortAttributes
+ var dsMulticastGemAttributeList []*tp_pb.GemPortAttributes
+ var dsUnicastGemAttributeList []*tp_pb.GemPortAttributes
+
+ if len(resInst.GemportIds) != int(tp.NumGemPorts) {
+ logger.Errorw(ctx, "mismatch-in-number-of-gemports-between-template-and-resource-instance",
+ log.Fields{"tpID": resInst.TpId, "totalResInstGemPortIDs": len(resInst.GemportIds), "totalTpTemplateGemPorts": tp.NumGemPorts})
+ return nil
+ }
+ for index := 0; index < int(tp.NumGemPorts); index++ {
+ usGemPortAttributeList = append(usGemPortAttributeList,
+ &tp_pb.GemPortAttributes{GemportId: resInst.GemportIds[index],
+ MaxQSize: tp.UpstreamGemPortAttributeList[index].MaxQSize,
+ PbitMap: tp.UpstreamGemPortAttributeList[index].PbitMap,
+ AesEncryption: tp.UpstreamGemPortAttributeList[index].AesEncryption,
+ SchedulingPolicy: tp.UpstreamGemPortAttributeList[index].SchedulingPolicy,
+ PriorityQ: tp.UpstreamGemPortAttributeList[index].PriorityQ,
+ Weight: tp.UpstreamGemPortAttributeList[index].Weight,
+ DiscardPolicy: tp.UpstreamGemPortAttributeList[index].DiscardPolicy,
+ DiscardConfig: tp.UpstreamGemPortAttributeList[index].DiscardConfig})
+ }
+
+ //put multicast and unicast downstream GEM port attributes in different lists first
+ for index := 0; index < len(tp.DownstreamGemPortAttributeList); index++ {
+ if isMulticastGem(tp.DownstreamGemPortAttributeList[index].IsMulticast) {
+ dsMulticastGemAttributeList = append(dsMulticastGemAttributeList,
+ &tp_pb.GemPortAttributes{
+ MulticastGemId: tp.DownstreamGemPortAttributeList[index].MulticastGemId,
+ MaxQSize: tp.DownstreamGemPortAttributeList[index].MaxQSize,
+ PbitMap: tp.DownstreamGemPortAttributeList[index].PbitMap,
+ AesEncryption: tp.DownstreamGemPortAttributeList[index].AesEncryption,
+ SchedulingPolicy: tp.DownstreamGemPortAttributeList[index].SchedulingPolicy,
+ PriorityQ: tp.DownstreamGemPortAttributeList[index].PriorityQ,
+ Weight: tp.DownstreamGemPortAttributeList[index].Weight,
+ DiscardPolicy: tp.DownstreamGemPortAttributeList[index].DiscardPolicy,
+ DiscardConfig: tp.DownstreamGemPortAttributeList[index].DiscardConfig,
+ IsMulticast: tp.DownstreamGemPortAttributeList[index].IsMulticast,
+ DynamicAccessControlList: tp.DownstreamGemPortAttributeList[index].DynamicAccessControlList,
+ StaticAccessControlList: tp.DownstreamGemPortAttributeList[index].StaticAccessControlList})
+ } else {
+ dsUnicastGemAttributeList = append(dsUnicastGemAttributeList,
+ &tp_pb.GemPortAttributes{
+ MaxQSize: tp.DownstreamGemPortAttributeList[index].MaxQSize,
+ PbitMap: tp.DownstreamGemPortAttributeList[index].PbitMap,
+ AesEncryption: tp.DownstreamGemPortAttributeList[index].AesEncryption,
+ SchedulingPolicy: tp.DownstreamGemPortAttributeList[index].SchedulingPolicy,
+ PriorityQ: tp.DownstreamGemPortAttributeList[index].PriorityQ,
+ Weight: tp.DownstreamGemPortAttributeList[index].Weight,
+ DiscardPolicy: tp.DownstreamGemPortAttributeList[index].DiscardPolicy,
+ DiscardConfig: tp.DownstreamGemPortAttributeList[index].DiscardConfig})
+ }
+ }
+ //add unicast downstream GEM ports to dsGemPortAttributeList
+ if dsUnicastGemAttributeList != nil {
+ for index := 0; index < int(tp.NumGemPorts); index++ {
+ dsGemPortAttributeList = append(dsGemPortAttributeList,
+ &tp_pb.GemPortAttributes{GemportId: resInst.GemportIds[index],
+ MaxQSize: dsUnicastGemAttributeList[index].MaxQSize,
+ PbitMap: dsUnicastGemAttributeList[index].PbitMap,
+ AesEncryption: dsUnicastGemAttributeList[index].AesEncryption,
+ SchedulingPolicy: dsUnicastGemAttributeList[index].SchedulingPolicy,
+ PriorityQ: dsUnicastGemAttributeList[index].PriorityQ,
+ 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 &tp_pb.TechProfileInstance{
+ SubscriberIdentifier: resInst.SubscriberIdentifier,
+ Name: tp.Name,
+ ProfileType: tp.ProfileType,
+ Version: tp.Version,
+ NumGemPorts: tp.NumGemPorts,
+ InstanceControl: tp.InstanceControl,
+ UsScheduler: &tp_pb.SchedulerAttributes{
+ AllocId: resInst.AllocId,
+ Direction: tp.UsScheduler.Direction,
+ AdditionalBw: tp.UsScheduler.AdditionalBw,
+ Priority: tp.UsScheduler.Priority,
+ Weight: tp.UsScheduler.Weight,
+ QSchedPolicy: tp.UsScheduler.QSchedPolicy},
+ DsScheduler: &tp_pb.SchedulerAttributes{
+ AllocId: resInst.AllocId,
+ Direction: tp.DsScheduler.Direction,
+ AdditionalBw: tp.DsScheduler.AdditionalBw,
+ Priority: tp.DsScheduler.Priority,
+ Weight: tp.DsScheduler.Weight,
+ QSchedPolicy: tp.DsScheduler.QSchedPolicy},
+ UpstreamGemPortAttributeList: usGemPortAttributeList,
+ DownstreamGemPortAttributeList: dsGemPortAttributeList}
+}
+
+// buildEponTpInstanceFromResourceInstance for EPON technology - build EponTpInstance from EponTechProfile template and ResourceInstance
+func (t *TechProfileMgr) buildEponTpInstanceFromResourceInstance(ctx context.Context, tp *tp_pb.EponTechProfile, resInst *tp_pb.ResourceInstance) *tp_pb.EponTechProfileInstance {
+
+ var usQueueAttributeList []*tp_pb.EPONQueueAttributes
+ var dsQueueAttributeList []*tp_pb.EPONQueueAttributes
+
+ if len(resInst.GemportIds) != int(tp.NumGemPorts) {
+ logger.Errorw(ctx, "mismatch-in-number-of-gemports-between-epon-tp-template-and-resource-instance",
+ log.Fields{"tpID": resInst.TpId, "totalResInstGemPortIDs": len(resInst.GemportIds), "totalTpTemplateGemPorts": tp.NumGemPorts})
+ return nil
+ }
+
+ for index := 0; index < int(tp.NumGemPorts); index++ {
+ usQueueAttributeList = append(usQueueAttributeList,
+ &tp_pb.EPONQueueAttributes{GemportId: resInst.GemportIds[index],
+ MaxQSize: tp.UpstreamQueueAttributeList[index].MaxQSize,
+ PbitMap: tp.UpstreamQueueAttributeList[index].PbitMap,
+ AesEncryption: tp.UpstreamQueueAttributeList[index].AesEncryption,
+ TrafficType: tp.UpstreamQueueAttributeList[index].TrafficType,
+ UnsolicitedGrantSize: tp.UpstreamQueueAttributeList[index].UnsolicitedGrantSize,
+ NominalInterval: tp.UpstreamQueueAttributeList[index].NominalInterval,
+ ToleratedPollJitter: tp.UpstreamQueueAttributeList[index].ToleratedPollJitter,
+ RequestTransmissionPolicy: tp.UpstreamQueueAttributeList[index].RequestTransmissionPolicy,
+ NumQSets: tp.UpstreamQueueAttributeList[index].NumQSets,
+ QThresholds: tp.UpstreamQueueAttributeList[index].QThresholds,
+ SchedulingPolicy: tp.UpstreamQueueAttributeList[index].SchedulingPolicy,
+ PriorityQ: tp.UpstreamQueueAttributeList[index].PriorityQ,
+ Weight: tp.UpstreamQueueAttributeList[index].Weight,
+ DiscardPolicy: tp.UpstreamQueueAttributeList[index].DiscardPolicy,
+ DiscardConfig: tp.UpstreamQueueAttributeList[index].DiscardConfig})
+ }
+
+ for index := 0; index < int(tp.NumGemPorts); index++ {
+ dsQueueAttributeList = append(dsQueueAttributeList,
+ &tp_pb.EPONQueueAttributes{GemportId: resInst.GemportIds[index],
+ MaxQSize: tp.DownstreamQueueAttributeList[index].MaxQSize,
+ PbitMap: tp.DownstreamQueueAttributeList[index].PbitMap,
+ AesEncryption: tp.DownstreamQueueAttributeList[index].AesEncryption,
+ SchedulingPolicy: tp.DownstreamQueueAttributeList[index].SchedulingPolicy,
+ PriorityQ: tp.DownstreamQueueAttributeList[index].PriorityQ,
+ Weight: tp.DownstreamQueueAttributeList[index].Weight,
+ DiscardPolicy: tp.DownstreamQueueAttributeList[index].DiscardPolicy,
+ DiscardConfig: tp.DownstreamQueueAttributeList[index].DiscardConfig})
+ }
+
+ return &tp_pb.EponTechProfileInstance{
+ SubscriberIdentifier: resInst.SubscriberIdentifier,
+ Name: tp.Name,
+ ProfileType: tp.ProfileType,
+ Version: tp.Version,
+ NumGemPorts: tp.NumGemPorts,
+ InstanceControl: tp.InstanceControl,
+ PackageType: tp.PackageType,
+ AllocId: resInst.AllocId,
+ UpstreamQueueAttributeList: usQueueAttributeList,
+ DownstreamQueueAttributeList: dsQueueAttributeList}
+}
+
+func (t *TechProfileMgr) getTpInstanceFromResourceInstance(ctx context.Context, resInst *tp_pb.ResourceInstance) *tp_pb.TechProfileInstance {
+ if resInst == nil {
+ logger.Error(ctx, "resource-instance-nil")
+ return nil
+ }
+ tp := t.getTPFromKVStore(ctx, resInst.TpId)
+ if tp == nil {
+ logger.Warnw(ctx, "tp-not-found-on-kv--creating-default-tp", log.Fields{"tpID": resInst.TpId})
+ tp = t.getDefaultTechProfile(ctx)
+ }
+ return t.buildTpInstanceFromResourceInstance(ctx, tp, resInst)
+}
+
+func (t *TechProfileMgr) getEponTpInstanceFromResourceInstance(ctx context.Context, resInst *tp_pb.ResourceInstance) *tp_pb.EponTechProfileInstance {
+ if resInst == nil {
+ logger.Error(ctx, "resource-instance-nil")
+ return nil
+ }
+ eponTp := t.getEponTPFromKVStore(ctx, resInst.TpId)
+ if eponTp == nil {
+ logger.Warnw(ctx, "tp-not-found-on-kv--creating-default-tp", log.Fields{"tpID": resInst.TpId})
+ eponTp = t.getDefaultEponProfile(ctx)
+ }
+ return t.buildEponTpInstanceFromResourceInstance(ctx, eponTp, resInst)
+}
+
+func (t *TechProfileMgr) reconcileTpInstancesToCache(ctx context.Context) error {
+
+ tech := t.resourceMgr.GetTechnology()
+ newCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
+ defer cancel()
+ kvPairs, _ := t.config.ResourceInstanceKVBacked.List(newCtx, tech)
+
+ if tech == xgspon || tech == xgpon || tech == gpon {
+ for keyPath, kvPair := range kvPairs {
+ logger.Debugw(ctx, "attempting-to-reconcile-tp-instance-from-resource-instance", log.Fields{"resourceInstPath": keyPath})
+ if value, err := kvstore.ToByte(kvPair.Value); err == nil {
+ var resInst tp_pb.ResourceInstance
+ if err = proto.Unmarshal(value, &resInst); err != nil {
+ logger.Errorw(ctx, "error-unmarshal-kv-pair", log.Fields{"err": err, "keyPath": keyPath, "value": value})
+ continue
+ } else {
+ if tpInst := t.getTpInstanceFromResourceInstance(ctx, &resInst); tpInst != nil {
+ // Trim the kv path by removing the default prefix part and get only the suffix part to reference the internal cache
+ keySuffixSlice := regexp.MustCompile(t.config.ResourceInstanceKVPathPrefix+"/").Split(keyPath, 2)
+ if len(keySuffixSlice) == 2 {
+ keySuffixFormatRegexp := regexp.MustCompile(`^[a-zA-Z\-]+/[0-9]+/olt-{[a-z0-9\-]+}/pon-{[0-9]+}/onu-{[0-9]+}/uni-{[0-9]+}$`)
+ // Make sure the keySuffixSlice is as per format [a-zA-Z-+]/[\d+]/olt-{[a-z0-9\-]+}/pon-{[0-9]+}/onu-{[0-9]+}/uni-{[0-9]+}
+ if !keySuffixFormatRegexp.Match([]byte(keySuffixSlice[1])) {
+ logger.Errorw(ctx, "kv-path-not-confirming-to-format", log.Fields{"kvPath": keySuffixSlice[1]})
+ continue
+ }
+ } else {
+ logger.Errorw(ctx, "kv-instance-key-path-not-in-the-expected-format", log.Fields{"kvPath": keyPath})
+ continue
+ }
+ t.tpInstanceMapLock.Lock()
+ t.tpInstanceMap[keySuffixSlice[1]] = tpInst
+ t.tpInstanceMapLock.Unlock()
+ logger.Debugw(ctx, "reconciled-tp-success", log.Fields{"keyPath": keyPath})
+ }
+ }
+ } else {
+ logger.Errorw(ctx, "error-converting-kv-pair-value-to-byte", log.Fields{"err": err})
+ }
+ }
+ } else if tech == epon {
+ for keyPath, kvPair := range kvPairs {
+ logger.Debugw(ctx, "attempting-to-reconcile-epon-tp-instance", log.Fields{"keyPath": keyPath})
+ if value, err := kvstore.ToByte(kvPair.Value); err == nil {
+ var resInst tp_pb.ResourceInstance
+ if err = proto.Unmarshal(value, &resInst); err != nil {
+ logger.Errorw(ctx, "error-unmarshal-kv-pair", log.Fields{"keyPath": keyPath, "value": value})
+ continue
+ } else {
+ if eponTpInst := t.getEponTpInstanceFromResourceInstance(ctx, &resInst); eponTpInst != nil {
+ // Trim the kv path by removing the default prefix part and get only the suffix part to reference the internal cache
+ keySuffixSlice := regexp.MustCompile(t.config.ResourceInstanceKVPathPrefix+"/").Split(keyPath, 2)
+ if len(keySuffixSlice) == 2 {
+ keySuffixFormatRegexp := regexp.MustCompile(`^[a-zA-Z\-]+/[0-9]+/olt-{[a-z0-9\-]+}/pon-{[0-9]+}/onu-{[0-9]+}/uni-{[0-9]+}$`)
+ // Make sure the keySuffixSlice is as per format [a-zA-Z-+]/[\d+]/olt-{[a-z0-9\-]+}/pon-{[0-9]+}/onu-{[0-9]+}/uni-{[0-9]+}
+ if !keySuffixFormatRegexp.Match([]byte(keySuffixSlice[1])) {
+ logger.Errorw(ctx, "kv-path-not-confirming-to-format", log.Fields{"kvPath": keySuffixSlice[1]})
+ continue
+ }
+ } else {
+ logger.Errorw(ctx, "kv-instance-key-path-not-in-the-expected-format", log.Fields{"kvPath": keyPath})
+ continue
+ }
+ t.epontpInstanceMapLock.Lock()
+ t.eponTpInstanceMap[keySuffixSlice[1]] = eponTpInst
+ t.epontpInstanceMapLock.Unlock()
+ logger.Debugw(ctx, "reconciled-epon-tp-success", log.Fields{"keyPath": keyPath})
+ }
+ }
+ } else {
+ logger.Errorw(ctx, "error-converting-kv-pair-value-to-byte", log.Fields{"err": err})
+ }
+ }
+ } else {
+ logger.Errorw(ctx, "unknown-tech", log.Fields{"tech": tech})
+ return fmt.Errorf("unknown-tech-%v", tech)
+ }
+
+ return nil
+}
diff --git a/vendor/github.com/opencord/voltha-lib-go/v7/pkg/techprofile/tech_profile_if.go b/vendor/github.com/opencord/voltha-lib-go/v7/pkg/techprofile/tech_profile_if.go
new file mode 100644
index 0000000..89a66b2
--- /dev/null
+++ b/vendor/github.com/opencord/voltha-lib-go/v7/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/v7/pkg/db"
+ tp_pb "github.com/opencord/voltha-protos/v5/go/tech_profile"
+)
+
+type TechProfileIf interface {
+ SetKVClient(ctx context.Context, pathPrefix string) *db.Backend
+ GetTechProfileInstanceKey(ctx context.Context, tpID uint32, uniPortName string) string
+ GetTPInstance(ctx context.Context, path string) (interface{}, error)
+ CreateTechProfileInstance(ctx context.Context, tpID uint32, uniPortName string, intfID uint32) (interface{}, error)
+ DeleteTechProfileInstance(ctx context.Context, tpID uint32, uniPortName string) error
+ GetUsScheduler(tpInstance *tp_pb.TechProfileInstance) *tp_pb.SchedulerConfig
+ GetDsScheduler(tpInstance *tp_pb.TechProfileInstance) *tp_pb.SchedulerConfig
+ GetTrafficScheduler(tpInstance *tp_pb.TechProfileInstance, SchedCfg *tp_pb.SchedulerConfig, ShapingCfg *tp_pb.TrafficShapingInfo) *tp_pb.TrafficScheduler
+ GetTrafficQueues(ctx context.Context, tp *tp_pb.TechProfileInstance, Dir tp_pb.Direction) ([]*tp_pb.TrafficQueue, error)
+ GetMulticastTrafficQueues(ctx context.Context, tp *tp_pb.TechProfileInstance) []*tp_pb.TrafficQueue
+ GetGemportForPbit(ctx context.Context, tp interface{}, Dir tp_pb.Direction, pbit uint32) interface{}
+ FindAllTpInstances(ctx context.Context, oltDeviceID string, tpID uint32, ponIntf uint32, onuID uint32) interface{}
+ GetResourceID(ctx context.Context, IntfID uint32, ResourceType string, NumIDs uint32) ([]uint32, error)
+ FreeResourceID(ctx context.Context, IntfID uint32, ResourceType string, ReleaseContent []uint32) error
+}