[VOL-4292] OpenOLT Adapter changes for gRPC migration
Change-Id: I5af2125f2c2f53ffc78c474a94314bba408f8bae
diff --git a/internal/pkg/config/config.go b/internal/pkg/config/config.go
index a08e73d..576ae72 100644
--- a/internal/pkg/config/config.go
+++ b/internal/pkg/config/config.go
@@ -27,16 +27,14 @@
const (
EtcdStoreName = "etcd"
defaultInstanceid = "openOlt001"
- defaultKafkaadapteraddress = "127.0.0.1:9092"
- defaultKafkaclusteraddress = "127.0.0.1:9094"
+ defaultKafkaclusteraddress = "127.0.0.1:9092"
defaultKvstoretype = EtcdStoreName
defaultKvstoretimeout = 5 * time.Second
+ defaultRPCTimeout = 10 * time.Second
defaultKvstoreaddress = "127.0.0.1:2379" // Port: Consul = 8500; Etcd = 2379
defaultLoglevel = "WARN"
defaultBanner = false
defaultDisplayVersionOnly = false
- defaultTopic = "openolt"
- defaultCoretopic = "rwcore"
defaultEventtopic = "voltha.events"
defaultOnunumber = 1
defaultProbeAddress = ":8080"
@@ -46,6 +44,8 @@
defaultHeartbeatCheckInterval = 15 * time.Second
// defaultHeartbeatFailReportInterval is the time adapter will wait before updating the state to the core.
defaultHeartbeatFailReportInterval = 0 * time.Second
+ defaultGrpcAddress = ":50060"
+ defaultCoreEndpoint = ":55555"
//defaultGrpcTimeoutInterval is the time in seconds a grpc call will wait before returning error.
defaultGrpcTimeoutInterval = 2 * time.Second
defaultCurrentReplica = 1
@@ -56,6 +56,9 @@
defaultOmccEncryption = false
defaultEnableONUStats = false
defaultEnableGEMStats = false
+ defaultMinBackoffRetryDelay = 500 * time.Millisecond
+ defaultMaxBackoffRetryDelay = 10 * time.Second
+ defaultAdapterEndpoint = "adapter-open-olt"
)
// AdapterFlags represents the set of configurations used by the read-write adaptercore service
@@ -63,13 +66,11 @@
// Command line parameters
AdapterName string
InstanceID string // NOTE what am I used for? why not cli but only ENV? TODO expose in the chart
- KafkaAdapterAddress string
KafkaClusterAddress string
KVStoreType string
KVStoreTimeout time.Duration
KVStoreAddress string
- Topic string
- CoreTopic string
+ RPCTimeout time.Duration
EventTopic string
LogLevel string
OnuNumber int
@@ -81,6 +82,8 @@
HeartbeatCheckInterval time.Duration
HeartbeatFailReportInterval time.Duration
GrpcTimeoutInterval time.Duration
+ GrpcAddress string
+ CoreEndpoint string
CurrentReplica int
TotalReplicas int
TraceEnabled bool
@@ -89,19 +92,19 @@
OmccEncryption bool
EnableONUStats bool
EnableGEMStats bool
+ MinBackoffRetryDelay time.Duration
+ MaxBackoffRetryDelay time.Duration
+ AdapterEndpoint string
}
// NewAdapterFlags returns a new RWCore config
func NewAdapterFlags() *AdapterFlags {
var adapterFlags = AdapterFlags{ // Default values
InstanceID: defaultInstanceid,
- KafkaAdapterAddress: defaultKafkaadapteraddress,
KafkaClusterAddress: defaultKafkaclusteraddress,
KVStoreType: defaultKvstoretype,
KVStoreTimeout: defaultKvstoretimeout,
KVStoreAddress: defaultKvstoreaddress,
- Topic: defaultTopic,
- CoreTopic: defaultCoretopic,
EventTopic: defaultEventtopic,
LogLevel: defaultLoglevel,
OnuNumber: defaultOnunumber,
@@ -112,6 +115,8 @@
NotLiveProbeInterval: defaultNotLiveProbeInterval,
HeartbeatCheckInterval: defaultHeartbeatCheckInterval,
HeartbeatFailReportInterval: defaultHeartbeatFailReportInterval,
+ GrpcAddress: defaultGrpcAddress,
+ CoreEndpoint: defaultCoreEndpoint,
GrpcTimeoutInterval: defaultGrpcTimeoutInterval,
TraceEnabled: defaultTraceEnabled,
TraceAgentAddress: defaultTraceAgentAddress,
@@ -119,6 +124,9 @@
OmccEncryption: defaultOmccEncryption,
EnableONUStats: defaultEnableONUStats,
EnableGEMStats: defaultEnableGEMStats,
+ RPCTimeout: defaultRPCTimeout,
+ MinBackoffRetryDelay: defaultMinBackoffRetryDelay,
+ MaxBackoffRetryDelay: defaultMaxBackoffRetryDelay,
}
return &adapterFlags
}
@@ -126,57 +134,150 @@
// ParseCommandArguments parses the arguments when running read-write adaptercore service
func (so *AdapterFlags) ParseCommandArguments() {
- flag.StringVar(&(so.KafkaAdapterAddress), "kafka_adapter_address", defaultKafkaadapteraddress, "Kafka - Adapter messaging address")
+ flag.StringVar(&(so.KafkaClusterAddress),
+ "kafka_cluster_address",
+ defaultKafkaclusteraddress,
+ "Kafka - Cluster messaging address")
- flag.StringVar(&(so.KafkaClusterAddress), "kafka_cluster_address", defaultKafkaclusteraddress, "Kafka - Cluster messaging address")
+ flag.StringVar(&(so.EventTopic),
+ "event_topic",
+ defaultEventtopic,
+ "Event topic")
- flag.StringVar(&(so.Topic), "adapter_topic", defaultTopic, "Open OLT topic")
+ flag.StringVar(&(so.KVStoreType),
+ "kv_store_type",
+ defaultKvstoretype,
+ "KV store type")
- flag.StringVar(&(so.CoreTopic), "core_topic", defaultCoretopic, "Core topic")
+ flag.DurationVar(&(so.KVStoreTimeout),
+ "kv_store_request_timeout",
+ defaultKvstoretimeout,
+ "The default timeout when making a kv store request")
- flag.StringVar(&(so.EventTopic), "event_topic", defaultEventtopic, "Event topic")
+ flag.StringVar(&(so.KVStoreAddress),
+ "kv_store_address",
+ defaultKvstoreaddress,
+ "KV store address")
- flag.StringVar(&(so.KVStoreType), "kv_store_type", defaultKvstoretype, "KV store type")
+ flag.StringVar(&(so.LogLevel),
+ "log_level",
+ defaultLoglevel,
+ "Log level")
- flag.DurationVar(&(so.KVStoreTimeout), "kv_store_request_timeout", defaultKvstoretimeout, "The default timeout when making a kv store request")
+ flag.IntVar(&(so.OnuNumber),
+ "onu_number",
+ defaultOnunumber,
+ "Number of ONUs")
- flag.StringVar(&(so.KVStoreAddress), "kv_store_address", defaultKvstoreaddress, "KV store address")
+ flag.BoolVar(&(so.Banner),
+ "banner",
+ defaultBanner,
+ "Show startup banner log lines")
- flag.StringVar(&(so.LogLevel), "log_level", defaultLoglevel, "Log level")
+ flag.BoolVar(&(so.DisplayVersionOnly),
+ "version",
+ defaultDisplayVersionOnly,
+ "Show version information and exit")
- flag.IntVar(&(so.OnuNumber), "onu_number", defaultOnunumber, "Number of ONUs")
+ flag.StringVar(&(so.ProbeAddress),
+ "probe_address",
+ defaultProbeAddress,
+ "The address on which to listen to answer liveness and readiness probe queries over HTTP.")
- flag.BoolVar(&(so.Banner), "banner", defaultBanner, "Show startup banner log lines")
+ flag.DurationVar(&(so.LiveProbeInterval),
+ "live_probe_interval",
+ defaultLiveProbeInterval,
+ "Number of seconds for the default liveliness check")
- flag.BoolVar(&(so.DisplayVersionOnly), "version", defaultDisplayVersionOnly, "Show version information and exit")
+ flag.DurationVar(&(so.NotLiveProbeInterval),
+ "not_live_probe_interval",
+ defaultNotLiveProbeInterval,
+ "Number of seconds for liveliness check if probe is not running")
- flag.StringVar(&(so.ProbeAddress), "probe_address", defaultProbeAddress, "The address on which to listen to answer liveness and readiness probe queries over HTTP.")
+ flag.DurationVar(&(so.HeartbeatCheckInterval),
+ "heartbeat_check_interval",
+ defaultHeartbeatCheckInterval,
+ "Number of seconds for heartbeat check interval")
- flag.DurationVar(&(so.LiveProbeInterval), "live_probe_interval", defaultLiveProbeInterval, "Number of seconds for the default liveliness check")
+ flag.DurationVar(&(so.HeartbeatFailReportInterval),
+ "heartbeat_fail_interval",
+ defaultHeartbeatFailReportInterval,
+ "Number of seconds adapter has to wait before reporting core on the heartbeat check failure")
- flag.DurationVar(&(so.NotLiveProbeInterval), "not_live_probe_interval", defaultNotLiveProbeInterval, "Number of seconds for liveliness check if probe is not running")
+ flag.DurationVar(&(so.GrpcTimeoutInterval),
+ "grpc_timeout_interval",
+ defaultGrpcTimeoutInterval,
+ "Number of seconds for GRPC timeout")
- flag.DurationVar(&(so.HeartbeatCheckInterval), "heartbeat_check_interval", defaultHeartbeatCheckInterval, "Number of seconds for heartbeat check interval")
+ flag.IntVar(&(so.CurrentReplica),
+ "current_replica",
+ defaultCurrentReplica,
+ "Replica number of this particular instance")
- flag.DurationVar(&(so.HeartbeatFailReportInterval), "heartbeat_fail_interval", defaultHeartbeatFailReportInterval, "Number of seconds adapter has to wait before reporting core on the heartbeat check failure")
+ flag.IntVar(&(so.TotalReplicas),
+ "total_replica",
+ defaultTotalReplicas,
+ "Total number of instances for this adapter")
- flag.DurationVar(&(so.GrpcTimeoutInterval), "grpc_timeout_interval", defaultGrpcTimeoutInterval, "Number of seconds for GRPC timeout")
+ flag.BoolVar(&(so.TraceEnabled),
+ "trace_enabled",
+ defaultTraceEnabled,
+ "Whether to send logs to tracing agent?")
- flag.IntVar(&(so.CurrentReplica), "current_replica", defaultCurrentReplica, "Replica number of this particular instance")
+ flag.StringVar(&(so.TraceAgentAddress),
+ "trace_agent_address",
+ defaultTraceAgentAddress,
+ "The address of tracing agent to which span info should be sent")
- flag.IntVar(&(so.TotalReplicas), "total_replica", defaultTotalReplicas, "Total number of instances for this adapter")
+ flag.BoolVar(&(so.LogCorrelationEnabled),
+ "log_correlation_enabled",
+ defaultLogCorrelationEnabled,
+ "Whether to enrich log statements with fields denoting operation being executed for achieving correlation?")
- flag.BoolVar(&(so.TraceEnabled), "trace_enabled", defaultTraceEnabled, "Whether to send logs to tracing agent?")
+ flag.BoolVar(&(so.OmccEncryption),
+ "omcc_encryption",
+ defaultOmccEncryption,
+ "OMCI Channel encryption status")
- flag.StringVar(&(so.TraceAgentAddress), "trace_agent_address", defaultTraceAgentAddress, "The address of tracing agent to which span info should be sent")
+ flag.BoolVar(&(so.EnableONUStats),
+ "enable_onu_stats",
+ defaultEnableONUStats,
+ "Enable ONU Statistics")
- flag.BoolVar(&(so.LogCorrelationEnabled), "log_correlation_enabled", defaultLogCorrelationEnabled, "Whether to enrich log statements with fields denoting operation being executed for achieving correlation?")
+ flag.BoolVar(&(so.EnableGEMStats),
+ "enable_gem_stats",
+ defaultEnableGEMStats,
+ "Enable GEM Statistics")
- flag.BoolVar(&(so.OmccEncryption), "omcc_encryption", defaultOmccEncryption, "OMCI Channel encryption status")
+ flag.StringVar(&(so.GrpcAddress),
+ "grpc_address",
+ defaultGrpcAddress,
+ "Adapter GRPC Server address")
- flag.BoolVar(&(so.EnableONUStats), "enable_onu_stats", defaultEnableONUStats, "Enable ONU Statistics")
+ flag.StringVar(&(so.CoreEndpoint),
+ "core_endpoint",
+ defaultCoreEndpoint,
+ "Core endpoint")
- flag.BoolVar(&(so.EnableGEMStats), "enable_gem_stats", defaultEnableGEMStats, "Enable GEM Statistics")
+ flag.StringVar(&(so.AdapterEndpoint),
+ "adapter_endpoint",
+ defaultAdapterEndpoint,
+ "Adapter Endpoint")
+
+ flag.DurationVar(&(so.RPCTimeout),
+ "rpc_timeout",
+ defaultRPCTimeout,
+ "The default timeout when making an RPC request")
+
+ flag.DurationVar(&(so.MinBackoffRetryDelay),
+ "min_retry_delay",
+ defaultMinBackoffRetryDelay,
+ "The minimum number of milliseconds to delay before a connection retry attempt")
+
+ flag.DurationVar(&(so.MaxBackoffRetryDelay),
+ "max_retry_delay",
+ defaultMaxBackoffRetryDelay,
+ "The maximum number of milliseconds to delay before a connection retry attempt")
flag.Parse()
containerName := getContainerInfo()
diff --git a/internal/pkg/config/config_test.go b/internal/pkg/config/config_test.go
index 332b9ea..e1199e1 100644
--- a/internal/pkg/config/config_test.go
+++ b/internal/pkg/config/config_test.go
@@ -17,9 +17,11 @@
//Package config provides the Log, kvstore, Kafka configuration
package config
-import "testing"
+import (
+ "testing"
+)
func TestAdapterFlags_ParseCommandArguments(t *testing.T) {
- so := NewAdapterFlags()
- so.ParseCommandArguments()
+ cf := &AdapterFlags{}
+ cf.ParseCommandArguments()
}
diff --git a/internal/pkg/core/common.go b/internal/pkg/core/common.go
index 28cf77e..f7df7a0 100644
--- a/internal/pkg/core/common.go
+++ b/internal/pkg/core/common.go
@@ -18,7 +18,7 @@
package core
import (
- "github.com/opencord/voltha-lib-go/v6/pkg/log"
+ "github.com/opencord/voltha-lib-go/v7/pkg/log"
)
var logger log.CLogger
diff --git a/internal/pkg/core/device_handler.go b/internal/pkg/core/device_handler.go
index 3dc62b4..c0e8cc8 100644
--- a/internal/pkg/core/device_handler.go
+++ b/internal/pkg/core/device_handler.go
@@ -30,26 +30,29 @@
"sync"
"time"
+ "github.com/golang/protobuf/ptypes/empty"
+ vgrpc "github.com/opencord/voltha-lib-go/v7/pkg/grpc"
+ "github.com/opencord/voltha-protos/v5/go/adapter_services"
+
"github.com/cenkalti/backoff/v3"
"github.com/gogo/protobuf/proto"
- "github.com/golang/protobuf/ptypes"
grpc_middleware "github.com/grpc-ecosystem/go-grpc-middleware"
grpc_opentracing "github.com/grpc-ecosystem/go-grpc-middleware/tracing/opentracing"
- "github.com/opencord/voltha-lib-go/v6/pkg/adapters/adapterif"
- "github.com/opencord/voltha-lib-go/v6/pkg/config"
- "github.com/opencord/voltha-lib-go/v6/pkg/events/eventif"
- flow_utils "github.com/opencord/voltha-lib-go/v6/pkg/flows"
- "github.com/opencord/voltha-lib-go/v6/pkg/log"
- "github.com/opencord/voltha-lib-go/v6/pkg/pmmetrics"
+ "github.com/opencord/voltha-lib-go/v7/pkg/config"
+ "github.com/opencord/voltha-lib-go/v7/pkg/events/eventif"
+ flow_utils "github.com/opencord/voltha-lib-go/v7/pkg/flows"
+ "github.com/opencord/voltha-lib-go/v7/pkg/log"
+ "github.com/opencord/voltha-lib-go/v7/pkg/pmmetrics"
+ conf "github.com/opencord/voltha-openolt-adapter/internal/pkg/config"
"github.com/opencord/voltha-openolt-adapter/internal/pkg/olterrors"
rsrcMgr "github.com/opencord/voltha-openolt-adapter/internal/pkg/resourcemanager"
- "github.com/opencord/voltha-protos/v4/go/common"
- "github.com/opencord/voltha-protos/v4/go/extension"
- ic "github.com/opencord/voltha-protos/v4/go/inter_container"
- of "github.com/opencord/voltha-protos/v4/go/openflow_13"
- oop "github.com/opencord/voltha-protos/v4/go/openolt"
- "github.com/opencord/voltha-protos/v4/go/voltha"
+ "github.com/opencord/voltha-protos/v5/go/common"
+ "github.com/opencord/voltha-protos/v5/go/extension"
+ ic "github.com/opencord/voltha-protos/v5/go/inter_container"
+ of "github.com/opencord/voltha-protos/v5/go/openflow_13"
+ oop "github.com/opencord/voltha-protos/v5/go/openolt"
+ "github.com/opencord/voltha-protos/v5/go/voltha"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
@@ -68,21 +71,23 @@
//DeviceHandler will interact with the OLT device.
type DeviceHandler struct {
- cm *config.ConfigManager
- device *voltha.Device
- coreProxy adapterif.CoreProxy
- AdapterProxy adapterif.AdapterProxy
- EventProxy eventif.EventProxy
- openOLT *OpenOLT
- exitChannel chan int
- lockDevice sync.RWMutex
- Client oop.OpenoltClient
- transitionMap *TransitionMap
- clientCon *grpc.ClientConn
- flowMgr []*OpenOltFlowMgr
- groupMgr *OpenOltGroupMgr
- eventMgr *OpenOltEventMgr
- resourceMgr []*rsrcMgr.OpenOltResourceMgr
+ cm *config.ConfigManager
+ device *voltha.Device
+ cfg *conf.AdapterFlags
+ coreClient *vgrpc.Client
+ childAdapterClients map[string]*vgrpc.Client
+ lockChildAdapterClients sync.RWMutex
+ EventProxy eventif.EventProxy
+ openOLT *OpenOLT
+ exitChannel chan int
+ lockDevice sync.RWMutex
+ Client oop.OpenoltClient
+ transitionMap *TransitionMap
+ clientCon *grpc.ClientConn
+ flowMgr []*OpenOltFlowMgr
+ groupMgr *OpenOltGroupMgr
+ eventMgr *OpenOltEventMgr
+ resourceMgr []*rsrcMgr.OpenOltResourceMgr
deviceInfo *oop.DeviceInfo
@@ -112,14 +117,15 @@
//OnuDevice represents ONU related info
type OnuDevice struct {
- deviceID string
- deviceType string
- serialNumber string
- onuID uint32
- intfID uint32
- proxyDeviceID string
- losRaised bool
- rdiRaised bool
+ deviceID string
+ deviceType string
+ serialNumber string
+ onuID uint32
+ intfID uint32
+ proxyDeviceID string
+ losRaised bool
+ rdiRaised bool
+ adapterEndpoint string
}
type onuIndicationMsg struct {
@@ -156,7 +162,7 @@
}
//NewOnuDevice creates a new Onu Device
-func NewOnuDevice(devID, deviceTp, serialNum string, onuID, intfID uint32, proxyDevID string, losRaised bool) *OnuDevice {
+func NewOnuDevice(devID, deviceTp, serialNum string, onuID, intfID uint32, proxyDevID string, losRaised bool, adapterEndpoint string) *OnuDevice {
var device OnuDevice
device.deviceID = devID
device.deviceType = deviceTp
@@ -165,15 +171,15 @@
device.intfID = intfID
device.proxyDeviceID = proxyDevID
device.losRaised = losRaised
+ device.adapterEndpoint = adapterEndpoint
return &device
}
//NewDeviceHandler creates a new device handler
-func NewDeviceHandler(cp adapterif.CoreProxy, ap adapterif.AdapterProxy, ep eventif.EventProxy, device *voltha.Device, adapter *OpenOLT, cm *config.ConfigManager) *DeviceHandler {
+func NewDeviceHandler(cc *vgrpc.Client, ep eventif.EventProxy, device *voltha.Device, adapter *OpenOLT, cm *config.ConfigManager, cfg *conf.AdapterFlags) *DeviceHandler {
var dh DeviceHandler
dh.cm = cm
- dh.coreProxy = cp
- dh.AdapterProxy = ap
+ dh.coreClient = cc
dh.EventProxy = ep
cloned := (proto.Clone(device)).(*voltha.Device)
dh.device = cloned
@@ -186,6 +192,8 @@
dh.activePorts = sync.Map{}
dh.stopIndications = make(chan bool, 1)
dh.perPonOnuIndicationChannel = make(map[uint32]onuIndicationChannels)
+ dh.childAdapterClients = make(map[string]*vgrpc.Client)
+ dh.cfg = cfg
// Create a slice of buffered channels for handling concurrent mcast flow/group.
dh.incomingMcastFlowOrGroup = make([]chan McastFlowOrGroupControlBlock, MaxNumOfGroupHandlerChannels)
dh.stopMcastHandlerRoutine = make([]chan bool, MaxNumOfGroupHandlerChannels)
@@ -219,6 +227,10 @@
defer dh.lockDevice.Unlock()
logger.Debug(ctx, "stopping-device-agent")
dh.exitChannel <- 1
+
+ // Stop the adapter grpc clients for that parent device
+ dh.deleteAdapterClients(ctx)
+
logger.Debug(ctx, "device-agent-stopped")
}
@@ -319,9 +331,19 @@
return olterrors.NewErrNotFound("port-label", log.Fields{"port-number": portNum, "port-type": portType}, err)
}
- if port, err := dh.coreProxy.GetDevicePort(log.WithSpanFromContext(context.TODO(), ctx), dh.device.Id, portNum); err == nil && port.Type == portType {
+ // Check if port exists
+ port, err := dh.getPortFromCore(ctx, &ic.PortFilter{
+ DeviceId: dh.device.Id,
+ Port: portNum,
+ })
+ if err == nil && port.Type == portType {
logger.Debug(ctx, "port-already-exists-updating-oper-status-of-port")
- if err := dh.coreProxy.PortStateUpdate(log.WithSpanFromContext(context.TODO(), ctx), dh.device.Id, portType, portNum, operStatus); err != nil {
+ err = dh.updatePortStateInCore(ctx, &ic.PortState{
+ DeviceId: dh.device.Id,
+ PortType: portType,
+ PortNo: portNum,
+ OperStatus: operStatus})
+ if err != nil {
return olterrors.NewErrAdapter("failed-to-update-port-state", log.Fields{
"device-id": dh.device.Id,
"port-type": portType,
@@ -330,9 +352,11 @@
}
return nil
}
+
// Now create Port
capacity := uint32(of.OfpPortFeatures_OFPPF_1GB_FD | of.OfpPortFeatures_OFPPF_FIBER)
- port := &voltha.Port{
+ port = &voltha.Port{
+ DeviceId: dh.device.Id,
PortNo: portNum,
Label: label,
Type: portType,
@@ -350,7 +374,8 @@
}
logger.Debugw(ctx, "sending-port-update-to-core", log.Fields{"port": port})
// Synchronous call to update device - this method is run in its own go routine
- if err := dh.coreProxy.PortCreated(log.WithSpanFromContext(context.TODO(), ctx), dh.device.Id, port); err != nil {
+ err = dh.createPortInCore(ctx, port)
+ if err != nil {
return olterrors.NewErrAdapter("error-creating-port", log.Fields{
"device-id": dh.device.Id,
"port-type": portType}, err)
@@ -360,7 +385,7 @@
}
func (dh *DeviceHandler) updateLocalDevice(ctx context.Context) {
- device, err := dh.coreProxy.GetDevice(log.WithSpanFromContext(context.TODO(), ctx), dh.device.Id, dh.device.Id)
+ device, err := dh.getDeviceFromCore(ctx, dh.device.Id)
if err != nil || device == nil {
logger.Errorf(ctx, "device-not-found", log.Fields{"device-id": dh.device.Id}, err)
return
@@ -658,9 +683,12 @@
}
// Synchronous call to update device state - this method is run in its own go routine
- if err := dh.coreProxy.DeviceStateUpdate(ctx, dh.device.Id, voltha.ConnectStatus_REACHABLE,
- voltha.OperStatus_ACTIVE); err != nil {
- return olterrors.NewErrAdapter("device-update-failed", log.Fields{"device-id": dh.device.Id}, err)
+ if err := dh.updateDeviceStateInCore(ctx, &ic.DeviceStateFilter{
+ DeviceId: dh.device.Id,
+ OperStatus: voltha.OperStatus_ACTIVE,
+ ConnStatus: voltha.ConnectStatus_REACHABLE,
+ }); err != nil {
+ return olterrors.NewErrAdapter("device-state-update-failed", log.Fields{"device-id": dh.device.Id}, err)
}
//Clear olt communication failure event
@@ -688,7 +716,7 @@
func (dh *DeviceHandler) doStateDown(ctx context.Context) error {
logger.Debugw(ctx, "do-state-down-start", log.Fields{"device-id": dh.device.Id})
- device, err := dh.coreProxy.GetDevice(ctx, dh.device.Id, dh.device.Id)
+ device, err := dh.getDeviceFromCore(ctx, dh.device.Id)
if err != nil || device == nil {
/*TODO: needs to handle error scenarios */
return olterrors.NewErrNotFound("device", log.Fields{"device-id": dh.device.Id}, err)
@@ -702,12 +730,16 @@
dh.device = cloned
dh.lockDevice.Unlock()
- if err = dh.coreProxy.DeviceStateUpdate(ctx, cloned.Id, cloned.ConnectStatus, cloned.OperStatus); err != nil {
+ if err = dh.updateDeviceStateInCore(ctx, &ic.DeviceStateFilter{
+ DeviceId: cloned.Id,
+ OperStatus: cloned.OperStatus,
+ ConnStatus: cloned.ConnectStatus,
+ }); err != nil {
return olterrors.NewErrAdapter("state-update-failed", log.Fields{"device-id": device.Id}, err)
}
//get the child device for the parent device
- onuDevices, err := dh.coreProxy.GetChildDevices(ctx, dh.device.Id)
+ onuDevices, err := dh.getChildDevicesFromCore(ctx, dh.device.Id)
if err != nil {
return olterrors.NewErrAdapter("child-device-fetch-failed", log.Fields{"device-id": dh.device.Id}, err)
}
@@ -715,11 +747,20 @@
// Update onu state as down in onu adapter
onuInd := oop.OnuIndication{}
onuInd.OperState = "down"
- err := dh.AdapterProxy.SendInterAdapterMessage(ctx, &onuInd, ic.InterAdapterMessageType_ONU_IND_REQUEST,
- dh.openOLT.config.Topic, onuDevice.Type, onuDevice.Id, onuDevice.ProxyAddress.DeviceId, "")
+
+ ogClient, err := dh.getChildAdapterServiceClient(onuDevice.AdapterEndpoint)
+ if err != nil {
+ return err
+ }
+ subCtx, cancel := context.WithTimeout(log.WithSpanFromContext(context.Background(), ctx), dh.cfg.RPCTimeout)
+ _, err = ogClient.OnuIndication(subCtx, &ic.OnuIndicationMessage{
+ DeviceId: onuDevice.Id,
+ OnuIndication: &onuInd,
+ })
+ cancel()
if err != nil {
_ = olterrors.NewErrCommunication("inter-adapter-send-failed", log.Fields{
- "source": dh.openOLT.config.Topic,
+ "source": dh.openOLT.config.AdapterEndpoint,
"onu-indicator": onuInd,
"device-type": onuDevice.Type,
"device-id": onuDevice.Id}, err).LogAt(log.ErrorLevel)
@@ -783,7 +824,7 @@
logger.Debugw(ctx, "olt-device-connected", log.Fields{"device-id": dh.device.Id})
// Case where OLT is disabled and then rebooted.
- device, err := dh.coreProxy.GetDevice(ctx, dh.device.Id, dh.device.Id)
+ device, err := dh.getDeviceFromCore(ctx, dh.device.Id)
if err != nil || device == nil {
/*TODO: needs to handle error scenarios */
return olterrors.NewErrAdapter("device-fetch-failed", log.Fields{"device-id": dh.device.Id}, err).LogAt(log.ErrorLevel)
@@ -795,7 +836,12 @@
cloned.ConnectStatus = voltha.ConnectStatus_REACHABLE
cloned.OperStatus = voltha.OperStatus_UNKNOWN
dh.device = cloned
- if err = dh.coreProxy.DeviceStateUpdate(ctx, cloned.Id, cloned.ConnectStatus, cloned.OperStatus); err != nil {
+
+ if err = dh.updateDeviceStateInCore(ctx, &ic.DeviceStateFilter{
+ DeviceId: cloned.Id,
+ OperStatus: cloned.OperStatus,
+ ConnStatus: cloned.ConnectStatus,
+ }); err != nil {
return olterrors.NewErrAdapter("device-state-update-failed", log.Fields{"device-id": dh.device.Id}, err).LogAt(log.ErrorLevel)
}
@@ -824,13 +870,13 @@
return nil
}
- ports, err := dh.coreProxy.ListDevicePorts(log.WithSpanFromContext(context.TODO(), ctx), dh.device.Id)
+ ports, err := dh.listDevicePortsFromCore(ctx, dh.device.Id)
if err != nil {
/*TODO: needs to handle error scenarios */
return olterrors.NewErrAdapter("fetch-ports-failed", log.Fields{"device-id": dh.device.Id}, err)
}
- dh.populateActivePorts(ctx, ports)
- if err := dh.disableAdminDownPorts(ctx, ports); err != nil {
+ dh.populateActivePorts(ctx, ports.Items)
+ if err := dh.disableAdminDownPorts(ctx, ports.Items); err != nil {
return olterrors.NewErrAdapter("port-status-update-failed", log.Fields{"ports": ports}, err)
}
@@ -932,7 +978,7 @@
}
// Synchronous call to update device - this method is run in its own go routine
- if err := dh.coreProxy.DeviceUpdate(log.WithSpanFromContext(context.TODO(), ctx), dh.device); err != nil {
+ if err = dh.updateDeviceInCore(ctx, dh.device); err != nil {
return nil, olterrors.NewErrAdapter("device-update-failed", log.Fields{"device-id": dh.device.Id}, err)
}
@@ -948,12 +994,12 @@
return
case <-time.After(time.Duration(dh.metrics.ToPmConfigs().DefaultFreq) * time.Second):
- ports, err := dh.coreProxy.ListDevicePorts(log.WithSpanFromContext(context.Background(), ctx), dh.device.Id)
+ ports, err := dh.listDevicePortsFromCore(ctx, dh.device.Id)
if err != nil {
logger.Warnw(ctx, "failed-to-list-ports", log.Fields{"device-id": dh.device.Id, "err": err})
continue
}
- for _, port := range ports {
+ for _, port := range ports.Items {
// NNI Stats
if port.Type == voltha.Port_ETHERNET_NNI {
intfID := PortNoToIntfID(port.PortNo, voltha.Port_ETHERNET_NNI)
@@ -989,7 +1035,14 @@
dh.transitionMap.Handle(ctx, DeviceInit)
// Now, set the initial PM configuration for that device
- if err := dh.coreProxy.DevicePMConfigUpdate(ctx, dh.metrics.ToPmConfigs()); err != nil {
+ cgClient, err := dh.coreClient.GetCoreServiceClient()
+ if err != nil {
+ logger.Errorw(ctx, "no-core-connection", log.Fields{"device-id": dh.device.Id, "error": err})
+ return
+ }
+
+ // Now, set the initial PM configuration for that device
+ if _, err := cgClient.DevicePMConfigUpdate(ctx, dh.metrics.ToPmConfigs()); err != nil {
_ = olterrors.NewErrAdapter("error-updating-performance-metrics", log.Fields{"device-id": device.Id}, err).LogAt(log.ErrorLevel)
}
}
@@ -1014,20 +1067,21 @@
}, nil
}
-// GetInterAdapterTechProfileDownloadMessage fetches the TechProfileDownloadMessage for the caller.
-func (dh *DeviceHandler) GetInterAdapterTechProfileDownloadMessage(ctx context.Context, tpPath string, ponPortNum uint32, onuID uint32, uniID uint32) *ic.InterAdapterTechProfileDownloadMessage {
- ifID, err := IntfIDFromPonPortNum(ctx, ponPortNum)
+// GetTechProfileDownloadMessage fetches the TechProfileDownloadMessage for the caller.
+func (dh *DeviceHandler) GetTechProfileDownloadMessage(ctx context.Context, request *ic.TechProfileInstanceRequestMessage) (*ic.TechProfileDownloadMessage, error) {
+ ifID, err := IntfIDFromPonPortNum(ctx, request.ParentPonPort)
if err != nil {
- return nil
+ return nil, err
}
- return dh.flowMgr[ifID].getTechProfileDownloadMessage(ctx, tpPath, ifID, onuID, uniID)
+ return dh.flowMgr[ifID].getTechProfileDownloadMessage(ctx, request.TpInstancePath, request.OnuId, request.DeviceId)
}
func (dh *DeviceHandler) omciIndication(ctx context.Context, omciInd *oop.OmciIndication) error {
- logger.Debugw(ctx, "omci-indication", log.Fields{"intf-id": omciInd.IntfId, "onu-id": omciInd.OnuId, "device-id": dh.device.Id})
+ logger.Debugw(ctx, "omci-indication", log.Fields{"intf-id": omciInd.IntfId, "onu-id": omciInd.OnuId, "parent-device-id": dh.device.Id})
var deviceType string
var deviceID string
var proxyDeviceID string
+ var childAdapterEndpoint string
transid := extractOmciTransactionID(omciInd.Pkt)
if logger.V(log.DebugLevel) {
@@ -1041,11 +1095,12 @@
logger.Debugw(ctx, "omci-indication-for-a-device-not-in-cache.", log.Fields{"intf-id": omciInd.IntfId, "onu-id": omciInd.OnuId, "device-id": dh.device.Id})
ponPort := IntfIDToPortNo(omciInd.GetIntfId(), voltha.Port_PON_OLT)
- kwargs := make(map[string]interface{})
- kwargs["onu_id"] = omciInd.OnuId
- kwargs["parent_port_no"] = ponPort
- onuDevice, err := dh.coreProxy.GetChildDevice(log.WithSpanFromContext(context.TODO(), ctx), dh.device.Id, kwargs)
+ onuDevice, err := dh.getChildDeviceFromCore(ctx, &ic.ChildDeviceFilter{
+ ParentId: dh.device.Id,
+ OnuId: omciInd.OnuId,
+ ParentPortNo: ponPort,
+ })
if err != nil {
return olterrors.NewErrNotFound("onu", log.Fields{
"intf-id": omciInd.IntfId,
@@ -1054,81 +1109,73 @@
deviceType = onuDevice.Type
deviceID = onuDevice.Id
proxyDeviceID = onuDevice.ProxyAddress.DeviceId
+ childAdapterEndpoint = onuDevice.AdapterEndpoint
//if not exist in cache, then add to cache.
- dh.onus.Store(onuKey, NewOnuDevice(deviceID, deviceType, onuDevice.SerialNumber, omciInd.OnuId, omciInd.IntfId, proxyDeviceID, false))
+ dh.onus.Store(onuKey, NewOnuDevice(deviceID, deviceType, onuDevice.SerialNumber, omciInd.OnuId, omciInd.IntfId, proxyDeviceID, false, onuDevice.AdapterEndpoint))
} else {
//found in cache
logger.Debugw(ctx, "omci-indication-for-a-device-in-cache.", log.Fields{"intf-id": omciInd.IntfId, "onu-id": omciInd.OnuId, "device-id": dh.device.Id})
deviceType = onuInCache.(*OnuDevice).deviceType
deviceID = onuInCache.(*OnuDevice).deviceID
proxyDeviceID = onuInCache.(*OnuDevice).proxyDeviceID
+ childAdapterEndpoint = onuInCache.(*OnuDevice).adapterEndpoint
}
- omciMsg := &ic.InterAdapterOmciMessage{Message: omciInd.Pkt}
- if err := dh.AdapterProxy.SendInterAdapterMessage(log.WithSpanFromContext(context.Background(), ctx), omciMsg,
- ic.InterAdapterMessageType_OMCI_RESPONSE, dh.openOLT.config.Topic, deviceType,
- deviceID, proxyDeviceID, ""); err != nil {
+ if err := dh.sendOmciIndicationToChildAdapter(ctx, childAdapterEndpoint, &ic.OmciMessage{
+ ParentDeviceId: proxyDeviceID,
+ ChildDeviceId: deviceID,
+ Message: omciInd.Pkt,
+ }); err != nil {
return olterrors.NewErrCommunication("omci-request", log.Fields{
- "source": dh.openOLT.config.Topic,
- "destination": deviceType,
+ "source": dh.openOLT.config.AdapterEndpoint,
+ "device-type": deviceType,
+ "destination": childAdapterEndpoint,
"onu-id": deviceID,
"proxy-device-id": proxyDeviceID}, err)
}
return nil
}
-//ProcessInterAdapterMessage sends the proxied messages to the target device
-// If the proxy address is not found in the unmarshalled message, it first fetches the onu device for which the message
-// is meant, and then send the unmarshalled omci message to this onu
-func (dh *DeviceHandler) ProcessInterAdapterMessage(ctx context.Context, msg *ic.InterAdapterMessage) error {
- logger.Debugw(ctx, "process-inter-adapter-message", log.Fields{"msgID": msg.Header.Id})
- if msg.Header.Type == ic.InterAdapterMessageType_OMCI_REQUEST {
- return dh.handleInterAdapterOmciMsg(ctx, msg)
- }
- return olterrors.NewErrInvalidValue(log.Fields{"inter-adapter-message-type": msg.Header.Type}, nil)
-}
+// //ProcessInterAdapterMessage sends the proxied messages to the target device
+// // If the proxy address is not found in the unmarshalled message, it first fetches the onu device for which the message
+// // is meant, and then send the unmarshalled omci message to this onu
+// func (dh *DeviceHandler) ProcessInterAdapterMessage(ctx context.Context, msg *ic.InterAdapterMessage) error {
+// logger.Debugw(ctx, "process-inter-adapter-message", log.Fields{"msgID": msg.Header.Id})
+// if msg.Header.Type == ic.InterAdapterMessageType_OMCI_REQUEST {
+// return dh.handleInterAdapterOmciMsg(ctx, msg)
+// }
+// return olterrors.NewErrInvalidValue(log.Fields{"inter-adapter-message-type": msg.Header.Type}, nil)
+// }
-func (dh *DeviceHandler) handleInterAdapterOmciMsg(ctx context.Context, msg *ic.InterAdapterMessage) error {
- msgID := msg.Header.Id
- fromTopic := msg.Header.FromTopic
- toTopic := msg.Header.ToTopic
- toDeviceID := msg.Header.ToDeviceId
- proxyDeviceID := msg.Header.ProxyDeviceId
-
- logger.Debugw(ctx, "omci-request-message-header", log.Fields{"msgID": msgID, "fromTopic": fromTopic, "toTopic": toTopic, "toDeviceID": toDeviceID, "proxyDeviceID": proxyDeviceID})
-
- msgBody := msg.GetBody()
-
- omciMsg := &ic.InterAdapterOmciMessage{}
- if err := ptypes.UnmarshalAny(msgBody, omciMsg); err != nil {
- return olterrors.NewErrAdapter("cannot-unmarshal-omci-msg-body", log.Fields{"msgbody": msgBody}, err)
- }
+// ProxyOmciMessage sends the proxied OMCI message to the target device
+func (dh *DeviceHandler) ProxyOmciMessage(ctx context.Context, omciMsg *ic.OmciMessage) error {
+ logger.Debugw(ctx, "proxy-omci-message", log.Fields{"parent-device-id": omciMsg.ParentDeviceId, "child-device-id": omciMsg.ChildDeviceId, "proxy-address": omciMsg.ProxyAddress, "connect-status": omciMsg.ConnectStatus})
if omciMsg.GetProxyAddress() == nil {
- onuDevice, err := dh.coreProxy.GetDevice(log.WithSpanFromContext(context.TODO(), ctx), dh.device.Id, toDeviceID)
+ onuDevice, err := dh.getDeviceFromCore(ctx, omciMsg.ChildDeviceId)
if err != nil {
return olterrors.NewErrNotFound("onu", log.Fields{
- "device-id": dh.device.Id,
- "onu-device-id": toDeviceID}, err)
+ "parent-device-id": dh.device.Id,
+ "child-device-id": omciMsg.ChildDeviceId}, err)
}
- logger.Debugw(ctx, "device-retrieved-from-core", log.Fields{"msgID": msgID, "fromTopic": fromTopic, "toTopic": toTopic, "toDeviceID": toDeviceID, "proxyDeviceID": proxyDeviceID})
- if err := dh.sendProxiedOmciMessage(ctx, onuDevice, omciMsg); err != nil {
+ logger.Debugw(ctx, "device-retrieved-from-core", log.Fields{"onu-device-proxy-address": onuDevice.ProxyAddress})
+ if err := dh.sendProxiedMessage(log.WithSpanFromContext(context.Background(), ctx), onuDevice, omciMsg); err != nil {
return olterrors.NewErrCommunication("send-failed", log.Fields{
- "device-id": dh.device.Id,
- "onu-device-id": toDeviceID}, err)
+ "parent-device-id": dh.device.Id,
+ "child-device-id": omciMsg.ChildDeviceId}, err)
}
} else {
- logger.Debugw(ctx, "proxy-address-found-in-omci-message", log.Fields{"msgID": msgID, "fromTopic": fromTopic, "toTopic": toTopic, "toDeviceID": toDeviceID, "proxyDeviceID": proxyDeviceID})
- if err := dh.sendProxiedOmciMessage(ctx, nil, omciMsg); err != nil {
+ logger.Debugw(ctx, "proxy-address-found-in-omci-message", log.Fields{"onu-device-proxy-address": omciMsg.ProxyAddress})
+ if err := dh.sendProxiedMessage(log.WithSpanFromContext(context.Background(), ctx), nil, omciMsg); err != nil {
return olterrors.NewErrCommunication("send-failed", log.Fields{
- "device-id": dh.device.Id,
- "onu-device-id": toDeviceID}, err)
+ "parent-device-id": dh.device.Id,
+ "child-device-id": omciMsg.ChildDeviceId}, err)
}
}
return nil
}
-func (dh *DeviceHandler) sendProxiedOmciMessage(ctx context.Context, onuDevice *voltha.Device, omciMsg *ic.InterAdapterOmciMessage) error {
+func (dh *DeviceHandler) sendProxiedMessage(ctx context.Context, onuDevice *voltha.Device, omciMsg *ic.OmciMessage) error {
var intfID uint32
var onuID uint32
var connectStatus common.ConnectStatus_Types
@@ -1200,13 +1247,6 @@
sn := dh.stringifySerialNumber(onuDiscInd.SerialNumber)
logger.Infow(ctx, "new-discovery-indication", log.Fields{"sn": sn})
- kwargs := make(map[string]interface{})
- if sn != "" {
- kwargs["serial_number"] = sn
- } else {
- return olterrors.NewErrInvalidValue(log.Fields{"serial-number": sn}, nil)
- }
-
var alarmInd oop.OnuAlarmIndication
raisedTs := time.Now().Unix()
if _, loaded := dh.discOnus.LoadOrStore(sn, true); loaded {
@@ -1244,7 +1284,10 @@
// check the ONU is already know to the OLT
// NOTE the second time the ONU is discovered this should return a device
- onuDevice, err := dh.coreProxy.GetChildDevice(ctx, dh.device.Id, kwargs)
+ onuDevice, err := dh.getChildDeviceFromCore(ctx, &ic.ChildDeviceFilter{
+ ParentId: dh.device.Id,
+ SerialNumber: sn,
+ })
if err != nil {
logger.Debugw(ctx, "core-proxy-get-child-device-failed", log.Fields{"parentDevice": dh.device.Id, "err": err, "sn": sn})
@@ -1280,8 +1323,14 @@
"serial-number": sn}, err)
}
- if onuDevice, err = dh.coreProxy.ChildDeviceDetected(log.WithSpanFromContext(context.TODO(), ctx), dh.device.Id, int(parentPortNo),
- "", int(channelID), string(onuDiscInd.SerialNumber.GetVendorId()), sn, int64(onuID)); err != nil {
+ if onuDevice, err = dh.sendChildDeviceDetectedToCore(ctx, &ic.DeviceDiscovery{
+ ParentId: dh.device.Id,
+ ParentPortNo: parentPortNo,
+ ChannelId: channelID,
+ VendorId: string(onuDiscInd.SerialNumber.GetVendorId()),
+ SerialNumber: sn,
+ OnuId: onuID,
+ }); err != nil {
dh.discOnus.Delete(sn)
dh.resourceMgr[ponintfid].FreeonuID(ctx, ponintfid, []uint32{onuID}) // NOTE I'm not sure this method is actually cleaning up the right thing
return olterrors.NewErrAdapter("core-proxy-child-device-detected-failed", log.Fields{
@@ -1298,6 +1347,14 @@
"device-id": dh.device.Id})
}
+ // Setup the gRPC connection to the adapter responsible for that onuDevice, if not setup yet
+ subCtx, cancel := context.WithTimeout(log.WithSpanFromContext(context.Background(), ctx), dh.cfg.RPCTimeout)
+ err = dh.setupChildInterAdapterClient(subCtx, onuDevice.AdapterEndpoint)
+ cancel()
+ if err != nil {
+ return olterrors.NewErrCommunication("no-connection-to-child-adapter", log.Fields{"device-id": onuDevice.Id}, err)
+ }
+
// we can now use the existing ONU Id
onuID = onuDevice.ProxyAddress.OnuId
//Insert the ONU into cache to use in OnuIndication.
@@ -1308,17 +1365,23 @@
"sn": sn})
onuKey := dh.formOnuKey(onuDiscInd.GetIntfId(), onuID)
- onuDev := NewOnuDevice(onuDevice.Id, onuDevice.Type, onuDevice.SerialNumber, onuID, onuDiscInd.GetIntfId(), onuDevice.ProxyAddress.DeviceId, false)
+ onuDev := NewOnuDevice(onuDevice.Id, onuDevice.Type, onuDevice.SerialNumber, onuID, onuDiscInd.GetIntfId(), onuDevice.ProxyAddress.DeviceId, false, onuDevice.AdapterEndpoint)
dh.onus.Store(onuKey, onuDev)
logger.Debugw(ctx, "new-onu-device-discovered",
log.Fields{"onu": onuDev,
"sn": sn})
- if err := dh.coreProxy.DeviceStateUpdate(ctx, onuDevice.Id, common.ConnectStatus_REACHABLE, common.OperStatus_DISCOVERED); err != nil {
+ if err := dh.updateDeviceStateInCore(ctx, &ic.DeviceStateFilter{
+ DeviceId: onuDevice.Id,
+ ParentDeviceId: dh.device.Id,
+ OperStatus: common.OperStatus_DISCOVERED,
+ ConnStatus: common.ConnectStatus_REACHABLE,
+ }); err != nil {
return olterrors.NewErrAdapter("failed-to-update-device-state", log.Fields{
"device-id": onuDevice.Id,
"serial-number": sn}, err)
}
+
logger.Infow(ctx, "onu-discovered-reachable", log.Fields{"device-id": onuDevice.Id, "sn": sn})
if err := dh.activateONU(ctx, onuDiscInd.IntfId, int64(onuID), onuDiscInd.SerialNumber, sn); err != nil {
return olterrors.NewErrAdapter("onu-activation-failed", log.Fields{
@@ -1330,7 +1393,6 @@
func (dh *DeviceHandler) onuIndication(ctx context.Context, onuInd *oop.OnuIndication) error {
- kwargs := make(map[string]interface{})
ponPort := IntfIDToPortNo(onuInd.GetIntfId(), voltha.Port_PON_OLT)
var onuDevice *voltha.Device
var err error
@@ -1349,19 +1411,21 @@
//If ONU id is discovered before then use GetDevice to get onuDevice because it is cheaper.
foundInCache = true
errFields["onu-id"] = onuInCache.(*OnuDevice).deviceID
- onuDevice, err = dh.coreProxy.GetDevice(ctx, dh.device.Id, onuInCache.(*OnuDevice).deviceID)
+ onuDevice, err = dh.getDeviceFromCore(ctx, onuInCache.(*OnuDevice).deviceID)
} else {
//If ONU not found in adapter cache then we have to use GetChildDevice to get onuDevice
if serialNumber != "" {
- kwargs["serial_number"] = serialNumber
errFields["serial-number"] = serialNumber
} else {
- kwargs["onu_id"] = onuInd.OnuId
- kwargs["parent_port_no"] = ponPort
errFields["onu-id"] = onuInd.OnuId
errFields["parent-port-no"] = ponPort
}
- onuDevice, err = dh.coreProxy.GetChildDevice(log.WithSpanFromContext(context.TODO(), ctx), dh.device.Id, kwargs)
+ onuDevice, err = dh.getChildDeviceFromCore(ctx, &ic.ChildDeviceFilter{
+ ParentId: dh.device.Id,
+ SerialNumber: serialNumber,
+ OnuId: onuInd.OnuId,
+ ParentPortNo: ponPort,
+ })
}
if err != nil || onuDevice == nil {
@@ -1383,7 +1447,7 @@
if !foundInCache {
onuKey := dh.formOnuKey(onuInd.GetIntfId(), onuInd.GetOnuId())
- dh.onus.Store(onuKey, NewOnuDevice(onuDevice.Id, onuDevice.Type, onuDevice.SerialNumber, onuInd.GetOnuId(), onuInd.GetIntfId(), onuDevice.ProxyAddress.DeviceId, false))
+ dh.onus.Store(onuKey, NewOnuDevice(onuDevice.Id, onuDevice.Type, onuDevice.SerialNumber, onuInd.GetOnuId(), onuInd.GetIntfId(), onuDevice.ProxyAddress.DeviceId, false, onuDevice.AdapterEndpoint))
}
if onuInd.OperState == "down" && onuInd.FailReason != oop.OnuIndication_ONU_ACTIVATION_FAIL_REASON_NONE {
@@ -1412,27 +1476,17 @@
}
switch onuInd.OperState {
- case "down":
+ case "up", "down":
logger.Debugw(ctx, "sending-interadapter-onu-indication", log.Fields{"onuIndication": onuInd, "device-id": onuDevice.Id, "operStatus": onuDevice.OperStatus, "adminStatus": onuDevice.AdminState})
- // TODO NEW CORE do not hardcode adapter name. Handler needs Adapter reference
- err := dh.AdapterProxy.SendInterAdapterMessage(ctx, onuInd, ic.InterAdapterMessageType_ONU_IND_REQUEST,
- dh.openOLT.config.Topic, onuDevice.Type, onuDevice.Id, onuDevice.ProxyAddress.DeviceId, "")
+
+ err := dh.sendOnuIndicationToChildAdapter(ctx, onuDevice.AdapterEndpoint, &ic.OnuIndicationMessage{
+ DeviceId: onuDevice.Id,
+ OnuIndication: onuInd,
+ })
if err != nil {
return olterrors.NewErrCommunication("inter-adapter-send-failed", log.Fields{
"onu-indicator": onuInd,
- "source": dh.openOLT.config.Topic,
- "device-type": onuDevice.Type,
- "device-id": onuDevice.Id}, err)
- }
- case "up":
- logger.Debugw(ctx, "sending-interadapter-onu-indication", log.Fields{"onuIndication": onuInd, "device-id": onuDevice.Id, "operStatus": onuDevice.OperStatus, "adminStatus": onuDevice.AdminState})
- // TODO NEW CORE do not hardcode adapter name. Handler needs Adapter reference
- err := dh.AdapterProxy.SendInterAdapterMessage(ctx, onuInd, ic.InterAdapterMessageType_ONU_IND_REQUEST,
- dh.openOLT.config.Topic, onuDevice.Type, onuDevice.Id, onuDevice.ProxyAddress.DeviceId, "")
- if err != nil {
- return olterrors.NewErrCommunication("inter-adapter-send-failed", log.Fields{
- "onu-indicator": onuInd,
- "source": dh.openOLT.config.Topic,
+ "source": dh.openOLT.config.AdapterEndpoint,
"device-type": onuDevice.Type,
"device-id": onuDevice.Id}, err)
}
@@ -1485,10 +1539,13 @@
log.Fields{"pon-port": parentPort,
"onu-id": onuID,
"device-id": dh.device.Id})
- kwargs := make(map[string]interface{})
- kwargs["onu_id"] = onuID
- kwargs["parent_port_no"] = parentPort
- onuDevice, err := dh.coreProxy.GetChildDevice(log.WithSpanFromContext(context.TODO(), ctx), dh.device.Id, kwargs)
+
+ onuDevice, err := dh.getChildDeviceFromCore(ctx, &ic.ChildDeviceFilter{
+ ParentId: dh.device.Id,
+ OnuId: onuID,
+ ParentPortNo: parentPort,
+ })
+
if err != nil {
return nil, olterrors.NewErrNotFound("onu-device", log.Fields{
"intf-id": parentPort,
@@ -1509,7 +1566,12 @@
"device-id": dh.device.Id,
})
}
- if err := dh.coreProxy.SendPacketIn(log.WithSpanFromContext(context.TODO(), ctx), dh.device.Id, logicalPort, packetPayload); err != nil {
+
+ if err := dh.sendPacketToCore(ctx, &ic.PacketIn{
+ DeviceId: dh.device.Id,
+ Port: logicalPort,
+ Packet: packetPayload,
+ }); err != nil {
return olterrors.NewErrCommunication("packet-send-failed", log.Fields{
"source": "adapter",
"destination": "core",
@@ -1646,8 +1708,13 @@
cloned := proto.Clone(device).(*voltha.Device)
//Update device Admin state
dh.device = cloned
+
// Update the all pon ports state on that device to disable and NNI remains active as NNI remains active in openolt agent.
- if err := dh.coreProxy.PortsStateUpdate(log.WithSpanFromContext(context.TODO(), ctx), cloned.Id, ^uint32(1<<voltha.Port_PON_OLT), voltha.OperStatus_UNKNOWN); err != nil {
+ if err := dh.updatePortsStateInCore(ctx, &ic.PortStateFilter{
+ DeviceId: cloned.Id,
+ PortTypeFilter: ^uint32(1 << voltha.Port_PON_OLT),
+ OperStatus: voltha.OperStatus_UNKNOWN,
+ }); err != nil {
return olterrors.NewErrAdapter("ports-state-update-failed", log.Fields{"device-id": device.Id}, err)
}
logger.Debugw(ctx, "disable-device-end", log.Fields{"device-id": device.Id})
@@ -1658,18 +1725,21 @@
// Update onu state as unreachable in onu adapter
onuInd := oop.OnuIndication{}
onuInd.OperState = state
+
//get the child device for the parent device
- onuDevices, err := dh.coreProxy.GetChildDevices(log.WithSpanFromContext(context.TODO(), ctx), dh.device.Id)
+ onuDevices, err := dh.getChildDevicesFromCore(ctx, dh.device.Id)
if err != nil {
logger.Errorw(ctx, "failed-to-get-child-devices-information", log.Fields{"device-id": dh.device.Id, "err": err})
}
if onuDevices != nil {
for _, onuDevice := range onuDevices.Items {
- err := dh.AdapterProxy.SendInterAdapterMessage(log.WithSpanFromContext(context.TODO(), ctx), &onuInd, ic.InterAdapterMessageType_ONU_IND_REQUEST,
- dh.openOLT.config.Topic, onuDevice.Type, onuDevice.Id, onuDevice.ProxyAddress.DeviceId, "")
+ err := dh.sendOnuIndicationToChildAdapter(ctx, onuDevice.AdapterEndpoint, &ic.OnuIndicationMessage{
+ DeviceId: onuDevice.Id,
+ OnuIndication: &onuInd,
+ })
if err != nil {
logger.Errorw(ctx, "failed-to-send-inter-adapter-message", log.Fields{"OnuInd": onuInd,
- "From Adapter": dh.openOLT.config.Topic, "DeviceType": onuDevice.Type, "device-id": onuDevice.Id})
+ "From Adapter": dh.openOLT.config.AdapterEndpoint, "DeviceType": onuDevice.Type, "device-id": onuDevice.Id})
}
}
@@ -1691,19 +1761,22 @@
logger.Debug(ctx, "olt-reenabled")
// Update the all ports state on that device to enable
-
- ports, err := dh.coreProxy.ListDevicePorts(ctx, device.Id)
+ ports, err := dh.listDevicePortsFromCore(ctx, device.Id)
if err != nil {
return olterrors.NewErrAdapter("list-ports-failed", log.Fields{"device-id": device.Id}, err)
}
- if err := dh.disableAdminDownPorts(ctx, ports); err != nil {
+ if err := dh.disableAdminDownPorts(ctx, ports.Items); err != nil {
return olterrors.NewErrAdapter("port-status-update-failed-after-olt-reenable", log.Fields{"device": device}, err)
}
//Update the device oper status as ACTIVE
device.OperStatus = voltha.OperStatus_ACTIVE
dh.device = device
- if err := dh.coreProxy.DeviceStateUpdate(log.WithSpanFromContext(context.TODO(), ctx), device.Id, device.ConnectStatus, device.OperStatus); err != nil {
+ if err := dh.updateDeviceStateInCore(ctx, &ic.DeviceStateFilter{
+ DeviceId: device.Id,
+ OperStatus: device.OperStatus,
+ ConnStatus: device.ConnectStatus,
+ }); err != nil {
return olterrors.NewErrAdapter("state-update-failed", log.Fields{
"device-id": device.Id,
"connect-status": device.ConnectStatus,
@@ -1858,7 +1931,11 @@
})
}
- if err := dh.coreProxy.SendPacketIn(log.WithSpanFromContext(context.TODO(), ctx), dh.device.Id, logicalPortNum, packetIn.Pkt); err != nil {
+ if err := dh.sendPacketToCore(ctx, &ic.PacketIn{
+ DeviceId: dh.device.Id,
+ Port: logicalPortNum,
+ Packet: packetIn.Pkt,
+ }); err != nil {
return olterrors.NewErrCommunication("send-packet-in", log.Fields{
"destination": "core",
"source": dh.device.Type,
@@ -1877,7 +1954,7 @@
}
// PacketOut sends packet-out from VOLTHA to OLT on the egress port provided
-func (dh *DeviceHandler) PacketOut(ctx context.Context, egressPortNo int, packet *of.OfpPacketOut) error {
+func (dh *DeviceHandler) PacketOut(ctx context.Context, egressPortNo uint32, packet *of.OfpPacketOut) error {
if logger.V(log.DebugLevel) {
logger.Debugw(ctx, "incoming-packet-out", log.Fields{
"device-id": dh.device.Id,
@@ -2033,7 +2110,7 @@
}
func (dh *DeviceHandler) updateStateUnreachable(ctx context.Context) {
- device, err := dh.coreProxy.GetDevice(ctx, dh.device.Id, dh.device.Id)
+ device, err := dh.getDeviceFromCore(ctx, dh.device.Id)
if err != nil || device == nil {
// One case where we have seen core returning an error for GetDevice call is after OLT device delete.
// After OLT delete, the adapter asks for OLT to reboot. When OLT is rebooted, shortly we loose heartbeat.
@@ -2048,18 +2125,29 @@
logger.Debugw(ctx, "update-state-unreachable", log.Fields{"device-id": dh.device.Id, "connect-status": device.ConnectStatus,
"admin-state": device.AdminState, "oper-status": device.OperStatus})
if device.ConnectStatus == voltha.ConnectStatus_REACHABLE {
- if err = dh.coreProxy.DeviceStateUpdate(ctx, dh.device.Id, voltha.ConnectStatus_UNREACHABLE, voltha.OperStatus_UNKNOWN); err != nil {
+ if err = dh.updateDeviceStateInCore(ctx, &ic.DeviceStateFilter{
+ DeviceId: dh.device.Id,
+ OperStatus: voltha.OperStatus_UNKNOWN,
+ ConnStatus: voltha.ConnectStatus_UNREACHABLE,
+ }); err != nil {
_ = olterrors.NewErrAdapter("device-state-update-failed", log.Fields{"device-id": dh.device.Id}, err).LogAt(log.ErrorLevel)
}
- if err = dh.coreProxy.PortsStateUpdate(ctx, dh.device.Id, 0, voltha.OperStatus_UNKNOWN); err != nil {
+
+ if err = dh.updatePortsStateInCore(ctx, &ic.PortStateFilter{
+ DeviceId: dh.device.Id,
+ PortTypeFilter: 0,
+ OperStatus: voltha.OperStatus_UNKNOWN,
+ }); err != nil {
_ = olterrors.NewErrAdapter("port-update-failed", log.Fields{"device-id": dh.device.Id}, err).Log()
}
//raise olt communication failure event
raisedTs := time.Now().Unix()
- device.ConnectStatus = voltha.ConnectStatus_UNREACHABLE
- device.OperStatus = voltha.OperStatus_UNKNOWN
- go dh.eventMgr.oltCommunicationEvent(ctx, device, raisedTs)
+ cloned := proto.Clone(device).(*voltha.Device)
+ cloned.ConnectStatus = voltha.ConnectStatus_UNREACHABLE
+ cloned.OperStatus = voltha.OperStatus_UNKNOWN
+ dh.device = cloned // update local copy of the device
+ go dh.eventMgr.oltCommunicationEvent(ctx, cloned, raisedTs)
dh.cleanupDeviceResources(ctx)
// Stop the Stats collector
@@ -2141,7 +2229,12 @@
dh.activePorts.Store(ponID, false)
logger.Infow(ctx, "disabled-pon-port", log.Fields{"out": out, "device-id": dh.device, "Port": port})
}
- if err := dh.coreProxy.PortStateUpdate(ctx, dh.device.Id, voltha.Port_PON_OLT, port.PortNo, operStatus); err != nil {
+ if err := dh.updatePortStateInCore(ctx, &ic.PortState{
+ DeviceId: dh.device.Id,
+ PortType: voltha.Port_PON_OLT,
+ PortNo: port.PortNo,
+ OperStatus: operStatus,
+ }); err != nil {
return olterrors.NewErrAdapter("port-state-update-failed", log.Fields{
"device-id": dh.device.Id,
"port": port.PortNo}, err)
@@ -2695,3 +2788,303 @@
return errResp(extension.GetValueResponse_ERROR, extension.GetValueResponse_REASON_UNDEFINED)
}
+
+/*
+Helper functions to communicate with Core
+*/
+
+func (dh *DeviceHandler) getDeviceFromCore(ctx context.Context, deviceID string) (*voltha.Device, error) {
+ cClient, err := dh.coreClient.GetCoreServiceClient()
+ if err != nil || cClient == nil {
+ return nil, err
+ }
+ subCtx, cancel := context.WithTimeout(log.WithSpanFromContext(context.Background(), ctx), dh.cfg.RPCTimeout)
+ defer cancel()
+ return cClient.GetDevice(subCtx, &common.ID{Id: deviceID})
+}
+
+func (dh *DeviceHandler) getChildDeviceFromCore(ctx context.Context, childDeviceFilter *ic.ChildDeviceFilter) (*voltha.Device, error) {
+ cClient, err := dh.coreClient.GetCoreServiceClient()
+ if err != nil || cClient == nil {
+ return nil, err
+ }
+ subCtx, cancel := context.WithTimeout(log.WithSpanFromContext(context.Background(), ctx), dh.cfg.RPCTimeout)
+ defer cancel()
+ return cClient.GetChildDevice(subCtx, childDeviceFilter)
+}
+
+func (dh *DeviceHandler) updateDeviceStateInCore(ctx context.Context, deviceStateFilter *ic.DeviceStateFilter) error {
+ cClient, err := dh.coreClient.GetCoreServiceClient()
+ if err != nil || cClient == nil {
+ return err
+ }
+ subCtx, cancel := context.WithTimeout(log.WithSpanFromContext(context.Background(), ctx), dh.cfg.RPCTimeout)
+ defer cancel()
+ _, err = cClient.DeviceStateUpdate(subCtx, deviceStateFilter)
+ return err
+}
+
+func (dh *DeviceHandler) getChildDevicesFromCore(ctx context.Context, deviceID string) (*voltha.Devices, error) {
+ cClient, err := dh.coreClient.GetCoreServiceClient()
+ if err != nil || cClient == nil {
+ return nil, err
+ }
+ subCtx, cancel := context.WithTimeout(log.WithSpanFromContext(context.Background(), ctx), dh.cfg.RPCTimeout)
+ defer cancel()
+ return cClient.GetChildDevices(subCtx, &common.ID{Id: deviceID})
+}
+
+func (dh *DeviceHandler) listDevicePortsFromCore(ctx context.Context, deviceID string) (*voltha.Ports, error) {
+ cClient, err := dh.coreClient.GetCoreServiceClient()
+ if err != nil || cClient == nil {
+ return nil, err
+ }
+ subCtx, cancel := context.WithTimeout(log.WithSpanFromContext(context.Background(), ctx), dh.cfg.RPCTimeout)
+ defer cancel()
+ return cClient.ListDevicePorts(subCtx, &common.ID{Id: deviceID})
+}
+
+func (dh *DeviceHandler) updateDeviceInCore(ctx context.Context, device *voltha.Device) error {
+ cClient, err := dh.coreClient.GetCoreServiceClient()
+ if err != nil || cClient == nil {
+ return err
+ }
+ subCtx, cancel := context.WithTimeout(log.WithSpanFromContext(context.Background(), ctx), dh.cfg.RPCTimeout)
+ defer cancel()
+ _, err = cClient.DeviceUpdate(subCtx, device)
+ return err
+}
+
+func (dh *DeviceHandler) sendChildDeviceDetectedToCore(ctx context.Context, deviceDiscoveryInfo *ic.DeviceDiscovery) (*voltha.Device, error) {
+ cClient, err := dh.coreClient.GetCoreServiceClient()
+ if err != nil || cClient == nil {
+ return nil, err
+ }
+ subCtx, cancel := context.WithTimeout(log.WithSpanFromContext(context.Background(), ctx), dh.cfg.RPCTimeout)
+ defer cancel()
+ return cClient.ChildDeviceDetected(subCtx, deviceDiscoveryInfo)
+}
+
+func (dh *DeviceHandler) sendPacketToCore(ctx context.Context, pkt *ic.PacketIn) error {
+ cClient, err := dh.coreClient.GetCoreServiceClient()
+ if err != nil || cClient == nil {
+ return err
+ }
+ subCtx, cancel := context.WithTimeout(log.WithSpanFromContext(context.Background(), ctx), dh.cfg.RPCTimeout)
+ defer cancel()
+ _, err = cClient.SendPacketIn(subCtx, pkt)
+ return err
+}
+
+func (dh *DeviceHandler) createPortInCore(ctx context.Context, port *voltha.Port) error {
+ cClient, err := dh.coreClient.GetCoreServiceClient()
+ if err != nil || cClient == nil {
+ return err
+ }
+ subCtx, cancel := context.WithTimeout(log.WithSpanFromContext(context.Background(), ctx), dh.cfg.RPCTimeout)
+ defer cancel()
+ _, err = cClient.PortCreated(subCtx, port)
+ return err
+}
+
+func (dh *DeviceHandler) updatePortsStateInCore(ctx context.Context, portFilter *ic.PortStateFilter) error {
+ cClient, err := dh.coreClient.GetCoreServiceClient()
+ if err != nil || cClient == nil {
+ return err
+ }
+ subCtx, cancel := context.WithTimeout(log.WithSpanFromContext(context.Background(), ctx), dh.cfg.RPCTimeout)
+ defer cancel()
+ _, err = cClient.PortsStateUpdate(subCtx, portFilter)
+ return err
+}
+
+func (dh *DeviceHandler) updatePortStateInCore(ctx context.Context, portState *ic.PortState) error {
+ cClient, err := dh.coreClient.GetCoreServiceClient()
+ if err != nil || cClient == nil {
+ return err
+ }
+ subCtx, cancel := context.WithTimeout(log.WithSpanFromContext(context.Background(), ctx), dh.cfg.RPCTimeout)
+ defer cancel()
+ _, err = cClient.PortStateUpdate(subCtx, portState)
+ return err
+}
+
+func (dh *DeviceHandler) getPortFromCore(ctx context.Context, portFilter *ic.PortFilter) (*voltha.Port, error) {
+ cClient, err := dh.coreClient.GetCoreServiceClient()
+ if err != nil || cClient == nil {
+ return nil, err
+ }
+ subCtx, cancel := context.WithTimeout(log.WithSpanFromContext(context.Background(), ctx), dh.cfg.RPCTimeout)
+ defer cancel()
+ return cClient.GetDevicePort(subCtx, portFilter)
+}
+
+/*
+Helper functions to communicate with child adapter
+*/
+
+func (dh *DeviceHandler) sendOmciIndicationToChildAdapter(ctx context.Context, childEndpoint string, response *ic.OmciMessage) error {
+ aClient, err := dh.getChildAdapterServiceClient(childEndpoint)
+ if err != nil || aClient == nil {
+ return err
+ }
+ logger.Debugw(ctx, "sending-omci-response", log.Fields{"response": response, "child-endpoint": childEndpoint})
+ subCtx, cancel := context.WithTimeout(log.WithSpanFromContext(context.Background(), ctx), dh.cfg.RPCTimeout)
+ defer cancel()
+ _, err = aClient.OmciIndication(subCtx, response)
+ return err
+}
+
+func (dh *DeviceHandler) sendOnuIndicationToChildAdapter(ctx context.Context, childEndpoint string, onuInd *ic.OnuIndicationMessage) error {
+ aClient, err := dh.getChildAdapterServiceClient(childEndpoint)
+ if err != nil || aClient == nil {
+ return err
+ }
+ logger.Debugw(ctx, "sending-onu-indication", log.Fields{"onu-indication": onuInd, "child-endpoint": childEndpoint})
+ subCtx, cancel := context.WithTimeout(log.WithSpanFromContext(context.Background(), ctx), dh.cfg.RPCTimeout)
+ defer cancel()
+ _, err = aClient.OnuIndication(subCtx, onuInd)
+ return err
+}
+
+func (dh *DeviceHandler) sendDeleteTContToChildAdapter(ctx context.Context, childEndpoint string, tContInfo *ic.DeleteTcontMessage) error {
+ aClient, err := dh.getChildAdapterServiceClient(childEndpoint)
+ if err != nil || aClient == nil {
+ return err
+ }
+ logger.Debugw(ctx, "sending-delete-tcont", log.Fields{"tcont": tContInfo, "child-endpoint": childEndpoint})
+ subCtx, cancel := context.WithTimeout(log.WithSpanFromContext(context.Background(), ctx), dh.cfg.RPCTimeout)
+ defer cancel()
+ _, err = aClient.DeleteTCont(subCtx, tContInfo)
+ return err
+}
+
+func (dh *DeviceHandler) sendDeleteGemPortToChildAdapter(ctx context.Context, childEndpoint string, gemPortInfo *ic.DeleteGemPortMessage) error {
+ aClient, err := dh.getChildAdapterServiceClient(childEndpoint)
+ if err != nil || aClient == nil {
+ return err
+ }
+ logger.Debugw(ctx, "sending-delete-gem-port", log.Fields{"gem-port-info": gemPortInfo, "child-endpoint": childEndpoint})
+ subCtx, cancel := context.WithTimeout(log.WithSpanFromContext(context.Background(), ctx), dh.cfg.RPCTimeout)
+ defer cancel()
+ _, err = aClient.DeleteGemPort(subCtx, gemPortInfo)
+ return err
+}
+
+func (dh *DeviceHandler) sendDownloadTechProfileToChildAdapter(ctx context.Context, childEndpoint string, tpDownloadInfo *ic.TechProfileDownloadMessage) error {
+ aClient, err := dh.getChildAdapterServiceClient(childEndpoint)
+ if err != nil || aClient == nil {
+ return err
+ }
+ logger.Debugw(ctx, "sending-tech-profile-download", log.Fields{"tp-download-info": tpDownloadInfo, "child-endpoint": childEndpoint})
+ subCtx, cancel := context.WithTimeout(log.WithSpanFromContext(context.Background(), ctx), dh.cfg.RPCTimeout)
+ defer cancel()
+ _, err = aClient.DownloadTechProfile(subCtx, tpDownloadInfo)
+ return err
+}
+
+/*
+Helper functions for remote communication
+*/
+
+// TODO: Use a connection tracker such that the adapter connection is stopped when the last device that adapter
+// supports is deleted
+func (dh *DeviceHandler) setupChildInterAdapterClient(ctx context.Context, endpoint string) error {
+ logger.Infow(ctx, "setting-child-adapter-connection", log.Fields{"child-endpoint": endpoint})
+
+ dh.lockChildAdapterClients.Lock()
+ defer dh.lockChildAdapterClients.Unlock()
+ if _, ok := dh.childAdapterClients[endpoint]; ok {
+ // Already set
+ return nil
+ }
+
+ // Setup child's adapter grpc connection
+ var err error
+ if dh.childAdapterClients[endpoint], err = vgrpc.NewClient(endpoint,
+ dh.onuAdapterRestarted,
+ vgrpc.ActivityCheck(true)); err != nil {
+ logger.Errorw(ctx, "grpc-client-not-created", log.Fields{"error": err, "endpoint": endpoint})
+ return err
+ }
+ go dh.childAdapterClients[endpoint].Start(log.WithSpanFromContext(context.TODO(), ctx), setAndTestAdapterServiceHandler)
+
+ // Wait until we have a connection to the child adapter.
+ // Unlimited retries or until context expires
+ subCtx := log.WithSpanFromContext(context.TODO(), ctx)
+ backoff := vgrpc.NewBackoff(dh.cfg.MinBackoffRetryDelay, dh.cfg.MaxBackoffRetryDelay, 0)
+ for {
+ client, err := dh.childAdapterClients[endpoint].GetOnuInterAdapterServiceClient()
+ if err == nil && client != nil {
+ logger.Infow(subCtx, "connected-to-child-adapter", log.Fields{"child-endpoint": endpoint})
+ break
+ }
+ logger.Warnw(subCtx, "connection-to-child-adapter-not-ready", log.Fields{"error": err, "child-endpoint": endpoint})
+ // Backoff
+ if err = backoff.Backoff(subCtx); err != nil {
+ logger.Errorw(subCtx, "received-error-on-backoff", log.Fields{"error": err, "child-endpoint": endpoint})
+ break
+ }
+ }
+ return nil
+}
+
+// func (dh *DeviceHandler) getChildAdapterServiceClient(endpoint string) (adapter_services.OnuInterAdapterServiceClient, error) {
+// dh.lockChildAdapterClients.RLock()
+// defer dh.lockChildAdapterClients.RUnlock()
+// if cgClient, ok := dh.childAdapterClients[endpoint]; ok {
+// return cgClient.GetOnuInterAdapterServiceClient()
+// }
+// return nil, fmt.Errorf("no-client-for-endpoint-%s", endpoint)
+// }
+
+func (dh *DeviceHandler) getChildAdapterServiceClient(endpoint string) (adapter_services.OnuInterAdapterServiceClient, error) {
+
+ // First check from cache
+ dh.lockChildAdapterClients.RLock()
+ if cgClient, ok := dh.childAdapterClients[endpoint]; ok {
+ dh.lockChildAdapterClients.RUnlock()
+ return cgClient.GetOnuInterAdapterServiceClient()
+ }
+ dh.lockChildAdapterClients.RUnlock()
+
+ // Set the child connection - can occur on restarts
+ ctx, cancel := context.WithTimeout(context.Background(), dh.cfg.RPCTimeout)
+ err := dh.setupChildInterAdapterClient(ctx, endpoint)
+ cancel()
+ if err != nil {
+ return nil, err
+ }
+
+ // Get the child client now
+ dh.lockChildAdapterClients.RLock()
+ defer dh.lockChildAdapterClients.RUnlock()
+ if cgClient, ok := dh.childAdapterClients[endpoint]; ok {
+ return cgClient.GetOnuInterAdapterServiceClient()
+ }
+ return nil, fmt.Errorf("no-client-for-endpoint-%s", endpoint)
+}
+
+func (dh *DeviceHandler) deleteAdapterClients(ctx context.Context) {
+ dh.lockChildAdapterClients.Lock()
+ defer dh.lockChildAdapterClients.Unlock()
+ for key, client := range dh.childAdapterClients {
+ client.Stop(ctx)
+ delete(dh.childAdapterClients, key)
+ }
+}
+
+// TODO: Any action the adapter needs to do following a onu adapter restart?
+func (dh *DeviceHandler) onuAdapterRestarted(ctx context.Context, endPoint string) error {
+ logger.Warnw(ctx, "onu-adapter-restarted", log.Fields{"endpoint": endPoint})
+ return nil
+}
+
+// setAndTestAdapterServiceHandler is used to test whether the remote gRPC service is up
+func setAndTestAdapterServiceHandler(ctx context.Context, conn *grpc.ClientConn) interface{} {
+ svc := adapter_services.NewOnuInterAdapterServiceClient(conn)
+ if h, err := svc.GetHealthStatus(ctx, &empty.Empty{}); err != nil || h.State != voltha.HealthStatus_HEALTHY {
+ return nil
+ }
+ return svc
+}
diff --git a/internal/pkg/core/device_handler_test.go b/internal/pkg/core/device_handler_test.go
index f021312..aa534d0 100644
--- a/internal/pkg/core/device_handler_test.go
+++ b/internal/pkg/core/device_handler_test.go
@@ -19,29 +19,28 @@
import (
"context"
- conf "github.com/opencord/voltha-lib-go/v6/pkg/config"
"net"
"reflect"
"sync"
"testing"
"time"
- "github.com/golang/protobuf/ptypes"
- "github.com/golang/protobuf/ptypes/any"
- "github.com/opencord/voltha-lib-go/v6/pkg/db"
- fu "github.com/opencord/voltha-lib-go/v6/pkg/flows"
- "github.com/opencord/voltha-lib-go/v6/pkg/log"
- "github.com/opencord/voltha-lib-go/v6/pkg/pmmetrics"
- ponrmgr "github.com/opencord/voltha-lib-go/v6/pkg/ponresourcemanager"
+ conf "github.com/opencord/voltha-lib-go/v7/pkg/config"
+ vgrpc "github.com/opencord/voltha-lib-go/v7/pkg/grpc"
+
+ "github.com/opencord/voltha-lib-go/v7/pkg/db"
+ fu "github.com/opencord/voltha-lib-go/v7/pkg/flows"
+ "github.com/opencord/voltha-lib-go/v7/pkg/pmmetrics"
+ ponrmgr "github.com/opencord/voltha-lib-go/v7/pkg/ponresourcemanager"
"github.com/opencord/voltha-openolt-adapter/internal/pkg/config"
"github.com/opencord/voltha-openolt-adapter/internal/pkg/olterrors"
"github.com/opencord/voltha-openolt-adapter/internal/pkg/resourcemanager"
"github.com/opencord/voltha-openolt-adapter/pkg/mocks"
- ic "github.com/opencord/voltha-protos/v4/go/inter_container"
- of "github.com/opencord/voltha-protos/v4/go/openflow_13"
- ofp "github.com/opencord/voltha-protos/v4/go/openflow_13"
- oop "github.com/opencord/voltha-protos/v4/go/openolt"
- "github.com/opencord/voltha-protos/v4/go/voltha"
+ ic "github.com/opencord/voltha-protos/v5/go/inter_container"
+ of "github.com/opencord/voltha-protos/v5/go/openflow_13"
+ ofp "github.com/opencord/voltha-protos/v5/go/openflow_13"
+ oop "github.com/opencord/voltha-protos/v5/go/openolt"
+ "github.com/opencord/voltha-protos/v5/go/voltha"
)
const (
@@ -56,17 +55,22 @@
FlowIDEnd = 10
)
-func newMockCoreProxy() *mocks.MockCoreProxy {
- mcp := mocks.MockCoreProxy{
+func newMockOnuInterAdapterService() *mocks.MockOnuInterAdapterService {
+ return &mocks.MockOnuInterAdapterService{}
+}
+
+func newMockCoreService() *mocks.MockCoreService {
+ mcp := mocks.MockCoreService{
Devices: make(map[string]*voltha.Device),
DevicePorts: make(map[string][]*voltha.Port),
}
var pm []*voltha.PmConfig
mcp.Devices["olt"] = &voltha.Device{
- Id: "olt",
- Root: true,
- ParentId: "logical_device",
- ParentPortNo: 1,
+ Id: "olt",
+ Root: true,
+ ParentId: "logical_device",
+ ParentPortNo: 1,
+ AdapterEndpoint: "mock-olt-endpoint",
ProxyAddress: &voltha.Device_ProxyAddress{
DeviceId: "olt",
DeviceType: "onu",
@@ -88,12 +92,12 @@
}
mcp.Devices["onu1"] = &voltha.Device{
- Id: "1",
- Root: false,
- ParentId: "olt",
- ParentPortNo: 1,
-
- OperStatus: 4,
+ Id: "1",
+ Root: false,
+ ParentId: "olt",
+ ParentPortNo: 1,
+ AdapterEndpoint: "mock-onu-endpoint",
+ OperStatus: 4,
ProxyAddress: &voltha.Device_ProxyAddress{
OnuId: 1,
ChannelId: 1,
@@ -114,12 +118,12 @@
}
mcp.Devices["onu2"] = &voltha.Device{
- Id: "2",
- Root: false,
- ParentId: "olt",
- OperStatus: 2,
-
- ParentPortNo: 1,
+ Id: "2",
+ Root: false,
+ ParentId: "olt",
+ OperStatus: 2,
+ AdapterEndpoint: "mock-onu-endpoint",
+ ParentPortNo: 1,
ProxyAddress: &voltha.Device_ProxyAddress{
OnuId: 2,
@@ -143,9 +147,10 @@
}
func newMockDeviceHandler() *DeviceHandler {
device := &voltha.Device{
- Id: "olt",
- Root: true,
- ParentId: "logical_device",
+ Id: "olt",
+ Root: true,
+ ParentId: "logical_device",
+ AdapterEndpoint: "mock-olt-endpoint",
ProxyAddress: &voltha.Device_ProxyAddress{
DeviceId: "olt",
DeviceType: "onu",
@@ -154,14 +159,14 @@
},
ConnectStatus: 1,
}
- cp := newMockCoreProxy()
- ap := &mocks.MockAdapterProxy{}
+ mcs := newMockCoreService()
+ cc := mocks.NewMockCoreClient(mcs)
ep := &mocks.MockEventProxy{}
cm := &conf.ConfigManager{}
cm.Backend = &db.Backend{StoreType: "etcd", Client: &mocks.MockKVClient{}}
cfg := &config.AdapterFlags{OmccEncryption: true}
- openOLT := &OpenOLT{coreProxy: cp, adapterProxy: ap, eventProxy: ep, config: cfg}
- dh := NewDeviceHandler(cp, ap, ep, device, openOLT, cm)
+ openOLT := &OpenOLT{eventProxy: ep, config: cfg}
+ dh := NewDeviceHandler(cc, ep, device, openOLT, cm, cfg)
oopRanges := []*oop.DeviceInfo_DeviceResourceRanges{{
IntfIds: []uint32{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15},
Technology: "xgs-pon",
@@ -170,6 +175,7 @@
deviceInf := &oop.DeviceInfo{Vendor: "openolt", Ranges: oopRanges, Model: "openolt", DeviceId: dh.device.Id, PonPorts: NumPonPorts}
dh.deviceInfo = deviceInf
+ dh.device = device
dh.resourceMgr = make([]*resourcemanager.OpenOltResourceMgr, deviceInf.PonPorts)
var i uint32
for i = 0; i < deviceInf.PonPorts; i++ {
@@ -264,6 +270,11 @@
}
dh.metrics = pmmetrics.NewPmMetrics(device.Id, pmmetrics.Frequency(2), pmmetrics.FrequencyOverride(false), pmmetrics.Grouped(false), pmmetrics.Metrics(pmNames))
+
+ // Set the children endpoints
+ dh.childAdapterClients = map[string]*vgrpc.Client{
+ "mock-onu-endpoint": mocks.NewMockChildAdapterClient(newMockOnuInterAdapterService()),
+ }
return dh
}
@@ -448,126 +459,126 @@
}
}
-func TestDeviceHandler_ProcessInterAdapterMessage(t *testing.T) {
- ctx := context.Background()
- dh := newMockDeviceHandler()
- proxyAddr := dh.device.ProxyAddress
- body := &ic.InterAdapterOmciMessage{
- Message: []byte("asdfasdfasdfasdfas"),
- ProxyAddress: proxyAddr,
- }
- body2 := &ic.InterAdapterOmciMessage{
- Message: []byte("asdfasdfasdfasdfas"),
- //ProxyAddress: &voltha.Device_ProxyAddress{},
- }
- body3 := &ic.InterAdapterTechProfileDownloadMessage{}
- var marshalledData *any.Any
- var err error
+// func TestDeviceHandler_ProcessInterAdapterMessage(t *testing.T) {
+// ctx := context.Background()
+// dh := newMockDeviceHandler()
+// proxyAddr := dh.device.ProxyAddress
+// body := &ic.InterAdapterOmciMessage{
+// Message: []byte("asdfasdfasdfasdfas"),
+// ProxyAddress: proxyAddr,
+// }
+// body2 := &ic.InterAdapterOmciMessage{
+// Message: []byte("asdfasdfasdfasdfas"),
+// //ProxyAddress: &voltha.Device_ProxyAddress{},
+// }
+// body3 := &ic.InterAdapterTechProfileDownloadMessage{}
+// var marshalledData *any.Any
+// var err error
- if marshalledData, err = ptypes.MarshalAny(body); err != nil {
- logger.Errorw(ctx, "cannot-marshal-request", log.Fields{"err": err})
- }
+// if marshalledData, err = ptypes.MarshalAny(body); err != nil {
+// logger.Errorw(ctx, "cannot-marshal-request", log.Fields{"err": err})
+// }
- var marshalledData1 *any.Any
+// var marshalledData1 *any.Any
- if marshalledData1, err = ptypes.MarshalAny(body2); err != nil {
- logger.Errorw(ctx, "cannot-marshal-request", log.Fields{"err": err})
- }
- var marshalledData2 *any.Any
+// if marshalledData1, err = ptypes.MarshalAny(body2); err != nil {
+// logger.Errorw(ctx, "cannot-marshal-request", log.Fields{"err": err})
+// }
+// var marshalledData2 *any.Any
- if marshalledData2, err = ptypes.MarshalAny(body3); err != nil {
- logger.Errorw(ctx, "cannot-marshal-request", log.Fields{"err": err})
- }
- type args struct {
- msg *ic.InterAdapterMessage
- }
- invalid := reflect.TypeOf(&olterrors.ErrInvalidValue{})
- tests := []struct {
- name string
- args args
- wantErr reflect.Type
- }{
- {"ProcessInterAdapterMessage-1", args{msg: &ic.InterAdapterMessage{
- Header: &ic.InterAdapterHeader{
- Id: "012345",
- Type: ic.InterAdapterMessageType_FLOW_REQUEST,
- },
- Body: marshalledData,
- }}, invalid},
- {"ProcessInterAdapterMessage-2", args{msg: &ic.InterAdapterMessage{
- Header: &ic.InterAdapterHeader{
- Id: "012345",
- Type: ic.InterAdapterMessageType_FLOW_RESPONSE,
- },
- Body: marshalledData1,
- }}, invalid},
- {"ProcessInterAdapterMessage-3", args{msg: &ic.InterAdapterMessage{
- Header: &ic.InterAdapterHeader{
- Id: "012345",
- Type: ic.InterAdapterMessageType_OMCI_REQUEST,
- },
- Body: marshalledData,
- }}, reflect.TypeOf(&olterrors.ErrCommunication{})},
- {"ProcessInterAdapterMessage-4", args{msg: &ic.InterAdapterMessage{
- Header: &ic.InterAdapterHeader{
- Id: "012345",
- Type: ic.InterAdapterMessageType_OMCI_RESPONSE,
- }, Body: marshalledData,
- }}, invalid},
- {"ProcessInterAdapterMessage-5", args{msg: &ic.InterAdapterMessage{
- Header: &ic.InterAdapterHeader{
- Id: "012345",
- Type: ic.InterAdapterMessageType_METRICS_REQUEST,
- }, Body: marshalledData1,
- }}, invalid},
- {"ProcessInterAdapterMessage-6", args{msg: &ic.InterAdapterMessage{
- Header: &ic.InterAdapterHeader{
- Id: "012345",
- Type: ic.InterAdapterMessageType_METRICS_RESPONSE,
- }, Body: marshalledData,
- }}, invalid},
- {"ProcessInterAdapterMessage-7", args{msg: &ic.InterAdapterMessage{
- Header: &ic.InterAdapterHeader{
- Id: "012345",
- Type: ic.InterAdapterMessageType_ONU_IND_REQUEST,
- }, Body: marshalledData,
- }}, invalid},
- {"ProcessInterAdapterMessage-8", args{msg: &ic.InterAdapterMessage{
- Header: &ic.InterAdapterHeader{
- Id: "012345",
- Type: ic.InterAdapterMessageType_ONU_IND_RESPONSE,
- }, Body: marshalledData,
- }}, invalid},
- {"ProcessInterAdapterMessage-9", args{msg: &ic.InterAdapterMessage{
- Header: &ic.InterAdapterHeader{
- Id: "012345",
- Type: ic.InterAdapterMessageType_TECH_PROFILE_DOWNLOAD_REQUEST,
- }, Body: marshalledData,
- }}, invalid},
- {"ProcessInterAdapterMessage-10", args{msg: &ic.InterAdapterMessage{
- Header: &ic.InterAdapterHeader{
- Id: "012345",
- Type: ic.InterAdapterMessageType_DELETE_GEM_PORT_REQUEST,
- }, Body: marshalledData2,
- }}, invalid},
- {"ProcessInterAdapterMessage-11", args{msg: &ic.InterAdapterMessage{
- Header: &ic.InterAdapterHeader{
- Id: "012345",
- Type: ic.InterAdapterMessageType_DELETE_TCONT_REQUEST,
- }, Body: marshalledData2,
- }}, invalid},
- }
- for _, tt := range tests {
- t.Run(tt.name, func(t *testing.T) {
+// if marshalledData2, err = ptypes.MarshalAny(body3); err != nil {
+// logger.Errorw(ctx, "cannot-marshal-request", log.Fields{"err": err})
+// }
+// type args struct {
+// msg *ic.InterAdapterMessage
+// }
+// invalid := reflect.TypeOf(&olterrors.ErrInvalidValue{})
+// tests := []struct {
+// name string
+// args args
+// wantErr reflect.Type
+// }{
+// {"ProcessInterAdapterMessage-1", args{msg: &ic.InterAdapterMessage{
+// Header: &ic.InterAdapterHeader{
+// Id: "012345",
+// Type: ic.InterAdapterMessageType_FLOW_REQUEST,
+// },
+// Body: marshalledData,
+// }}, invalid},
+// {"ProcessInterAdapterMessage-2", args{msg: &ic.InterAdapterMessage{
+// Header: &ic.InterAdapterHeader{
+// Id: "012345",
+// Type: ic.InterAdapterMessageType_FLOW_RESPONSE,
+// },
+// Body: marshalledData1,
+// }}, invalid},
+// {"ProcessInterAdapterMessage-3", args{msg: &ic.InterAdapterMessage{
+// Header: &ic.InterAdapterHeader{
+// Id: "012345",
+// Type: ic.InterAdapterMessageType_OMCI_REQUEST,
+// },
+// Body: marshalledData,
+// }}, reflect.TypeOf(&olterrors.ErrCommunication{})},
+// {"ProcessInterAdapterMessage-4", args{msg: &ic.InterAdapterMessage{
+// Header: &ic.InterAdapterHeader{
+// Id: "012345",
+// Type: ic.InterAdapterMessageType_OMCI_RESPONSE,
+// }, Body: marshalledData,
+// }}, invalid},
+// {"ProcessInterAdapterMessage-5", args{msg: &ic.InterAdapterMessage{
+// Header: &ic.InterAdapterHeader{
+// Id: "012345",
+// Type: ic.InterAdapterMessageType_METRICS_REQUEST,
+// }, Body: marshalledData1,
+// }}, invalid},
+// {"ProcessInterAdapterMessage-6", args{msg: &ic.InterAdapterMessage{
+// Header: &ic.InterAdapterHeader{
+// Id: "012345",
+// Type: ic.InterAdapterMessageType_METRICS_RESPONSE,
+// }, Body: marshalledData,
+// }}, invalid},
+// {"ProcessInterAdapterMessage-7", args{msg: &ic.InterAdapterMessage{
+// Header: &ic.InterAdapterHeader{
+// Id: "012345",
+// Type: ic.InterAdapterMessageType_ONU_IND_REQUEST,
+// }, Body: marshalledData,
+// }}, invalid},
+// {"ProcessInterAdapterMessage-8", args{msg: &ic.InterAdapterMessage{
+// Header: &ic.InterAdapterHeader{
+// Id: "012345",
+// Type: ic.InterAdapterMessageType_ONU_IND_RESPONSE,
+// }, Body: marshalledData,
+// }}, invalid},
+// {"ProcessInterAdapterMessage-9", args{msg: &ic.InterAdapterMessage{
+// Header: &ic.InterAdapterHeader{
+// Id: "012345",
+// Type: ic.InterAdapterMessageType_TECH_PROFILE_DOWNLOAD_REQUEST,
+// }, Body: marshalledData,
+// }}, invalid},
+// {"ProcessInterAdapterMessage-10", args{msg: &ic.InterAdapterMessage{
+// Header: &ic.InterAdapterHeader{
+// Id: "012345",
+// Type: ic.InterAdapterMessageType_DELETE_GEM_PORT_REQUEST,
+// }, Body: marshalledData2,
+// }}, invalid},
+// {"ProcessInterAdapterMessage-11", args{msg: &ic.InterAdapterMessage{
+// Header: &ic.InterAdapterHeader{
+// Id: "012345",
+// Type: ic.InterAdapterMessageType_DELETE_TCONT_REQUEST,
+// }, Body: marshalledData2,
+// }}, invalid},
+// }
+// for _, tt := range tests {
+// t.Run(tt.name, func(t *testing.T) {
- if err := dh.ProcessInterAdapterMessage(ctx, tt.args.msg); reflect.TypeOf(err) != tt.wantErr {
- t.Errorf("DeviceHandler.ProcessInterAdapterMessage() error = %v, wantErr %v", err, tt.wantErr)
- }
- })
- }
-}
+// if err := dh.ProcessInterAdapterMessage(ctx, tt.args.msg); reflect.TypeOf(err) != tt.wantErr {
+// t.Errorf("DeviceHandler.ProcessInterAdapterMessage() error = %v, wantErr %v", err, tt.wantErr)
+// }
+// })
+// }
+// }
-func TestDeviceHandler_sendProxiedMessage(t *testing.T) {
+func TestDeviceHandler_ProxyOmciMessage(t *testing.T) {
ctx := context.Background()
dh1 := newMockDeviceHandler()
dh2 := negativeDeviceHandler()
@@ -585,17 +596,16 @@
}
device2 := device1
device2.ConnectStatus = 2
- iaomciMsg1 := &ic.InterAdapterOmciMessage{
+ iaomciMsg1 := &ic.OmciMessage{
ProxyAddress: &voltha.Device_ProxyAddress{
DeviceId: "onu2",
DeviceType: "onu",
ChannelId: 1,
ChannelGroupId: 1,
- //OnuId: 2,
},
ConnectStatus: 1,
}
- iaomciMsg2 := &ic.InterAdapterOmciMessage{
+ iaomciMsg2 := &ic.OmciMessage{
ProxyAddress: &voltha.Device_ProxyAddress{
DeviceId: "onu3",
DeviceType: "onu",
@@ -606,23 +616,23 @@
}
type args struct {
onuDevice *voltha.Device
- omciMsg *ic.InterAdapterOmciMessage
+ omciMsg *ic.OmciMessage
}
tests := []struct {
name string
devicehandler *DeviceHandler
args args
}{
- {"sendProxiedMessage-1", dh1, args{onuDevice: device1, omciMsg: &ic.InterAdapterOmciMessage{}}},
- {"sendProxiedMessage-2", dh1, args{onuDevice: device2, omciMsg: &ic.InterAdapterOmciMessage{}}},
+ {"sendProxiedMessage-1", dh1, args{onuDevice: device1, omciMsg: &ic.OmciMessage{}}},
+ {"sendProxiedMessage-2", dh1, args{onuDevice: device2, omciMsg: &ic.OmciMessage{}}},
{"sendProxiedMessage-3", dh1, args{onuDevice: nil, omciMsg: iaomciMsg1}},
{"sendProxiedMessage-4", dh1, args{onuDevice: nil, omciMsg: iaomciMsg2}},
{"sendProxiedMessage-5", dh2, args{onuDevice: nil, omciMsg: iaomciMsg2}},
- {"sendProxiedMessage-6", dh2, args{onuDevice: device1, omciMsg: &ic.InterAdapterOmciMessage{}}},
+ {"sendProxiedMessage-6", dh2, args{onuDevice: device1, omciMsg: &ic.OmciMessage{}}},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
- _ = tt.devicehandler.sendProxiedOmciMessage(ctx, tt.args.onuDevice, tt.args.omciMsg)
+ _ = tt.devicehandler.ProxyOmciMessage(ctx, tt.args.omciMsg)
//TODO: actually verify test cases
})
}
@@ -735,8 +745,8 @@
dh2 := negativeDeviceHandler()
dh3 := newMockDeviceHandler()
dh3.onus = sync.Map{}
- dh3.onus.Store("onu1", NewOnuDevice("onu1", "onu1", "onu1", 1, 1, "onu1", false))
- dh3.onus.Store("onu2", NewOnuDevice("onu2", "onu2", "onu2", 2, 2, "onu2", false))
+ dh3.onus.Store("onu1", NewOnuDevice("onu1", "onu1", "onu1", 1, 1, "onu1", false, "mock_endpoint"))
+ dh3.onus.Store("onu2", NewOnuDevice("onu2", "onu2", "onu2", 2, 2, "onu2", false, "mock_endpoint"))
type args struct {
indication *oop.Indication
@@ -794,6 +804,7 @@
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
dh := tt.deviceHandler
+ time.Sleep(5 * time.Millisecond)
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
dh.handleIndication(ctx, tt.args.indication)
@@ -994,7 +1005,7 @@
dh := tt.devicehandler
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
- if err := dh.PacketOut(ctx, tt.args.egressPortNo, tt.args.packet); (err != nil) != tt.wantErr {
+ if err := dh.PacketOut(ctx, uint32(tt.args.egressPortNo), tt.args.packet); (err != nil) != tt.wantErr {
t.Errorf("DeviceHandler.PacketOut() error = %v, wantErr %v", err, tt.wantErr)
}
})
@@ -1093,8 +1104,8 @@
dh1.discOnus.Store("onu3", true)
dh1.discOnus.Store("onu4", true)
dh1.onus = sync.Map{}
- dh1.onus.Store("onu3", NewOnuDevice("onu3", "onu3", "onu3", 3, 3, "onu3", true))
- dh1.onus.Store("onu4", NewOnuDevice("onu4", "onu4", "onu4", 4, 4, "onu4", true))
+ dh1.onus.Store("onu3", NewOnuDevice("onu3", "onu3", "onu3", 3, 3, "onu3", true, "mock_endpoint"))
+ dh1.onus.Store("onu4", NewOnuDevice("onu4", "onu4", "onu4", 4, 4, "onu4", true, "mock_endpoint"))
dh2 := negativeDeviceHandler()
type args struct {
onuDiscInd *oop.OnuDiscIndication
@@ -1190,11 +1201,13 @@
dh *DeviceHandler
}
dh := newMockDeviceHandler()
- dh.coreProxy.(*mocks.MockCoreProxy).DevicePorts[dh.device.Id] = []*voltha.Port{
+ mcs := newMockCoreService()
+ mcs.DevicePorts[dh.device.Id] = []*voltha.Port{
{PortNo: 1, Label: "pon", Type: voltha.Port_PON_OLT},
{PortNo: 1048577, Label: "nni", Type: voltha.Port_ETHERNET_NNI},
{PortNo: 1048578, Label: "nni", Type: voltha.Port_ETHERNET_NNI},
}
+ dh.coreClient.SetService(mcs)
dh.portStats.NorthBoundPort = make(map[uint32]*NniPort)
dh.portStats.NorthBoundPort[1] = &NniPort{Name: "OLT-1"}
dh.portStats.NorthBoundPort[2] = &NniPort{Name: "OLT-1"}
@@ -1204,7 +1217,9 @@
dh.portStats.SouthBoundPort[uint32(i)] = &PonPort{DeviceID: "OLT-1"}
}
dh1 := newMockDeviceHandler()
- dh1.coreProxy.(*mocks.MockCoreProxy).DevicePorts[dh.device.Id] = []*voltha.Port{}
+ mcs = newMockCoreService()
+ mcs.DevicePorts[dh.device.Id] = []*voltha.Port{}
+ dh.coreClient.SetService(mcs)
tests := []struct {
name string
args args
diff --git a/internal/pkg/core/olt_platform.go b/internal/pkg/core/olt_platform.go
index 1bacc64..f9b735f 100644
--- a/internal/pkg/core/olt_platform.go
+++ b/internal/pkg/core/olt_platform.go
@@ -20,11 +20,11 @@
import (
"context"
- "github.com/opencord/voltha-lib-go/v6/pkg/flows"
- "github.com/opencord/voltha-lib-go/v6/pkg/log"
+ "github.com/opencord/voltha-lib-go/v7/pkg/flows"
+ "github.com/opencord/voltha-lib-go/v7/pkg/log"
"github.com/opencord/voltha-openolt-adapter/internal/pkg/olterrors"
- ofp "github.com/opencord/voltha-protos/v4/go/openflow_13"
- "github.com/opencord/voltha-protos/v4/go/voltha"
+ ofp "github.com/opencord/voltha-protos/v5/go/openflow_13"
+ "github.com/opencord/voltha-protos/v5/go/voltha"
)
/*=====================================================================
diff --git a/internal/pkg/core/olt_platform_test.go b/internal/pkg/core/olt_platform_test.go
index 9778417..c05282a 100644
--- a/internal/pkg/core/olt_platform_test.go
+++ b/internal/pkg/core/olt_platform_test.go
@@ -19,13 +19,14 @@
import (
"context"
- fu "github.com/opencord/voltha-lib-go/v6/pkg/flows"
- "github.com/opencord/voltha-openolt-adapter/internal/pkg/olterrors"
- ofp "github.com/opencord/voltha-protos/v4/go/openflow_13"
- "github.com/opencord/voltha-protos/v4/go/voltha"
"math"
"reflect"
"testing"
+
+ fu "github.com/opencord/voltha-lib-go/v7/pkg/flows"
+ "github.com/opencord/voltha-openolt-adapter/internal/pkg/olterrors"
+ ofp "github.com/opencord/voltha-protos/v5/go/openflow_13"
+ "github.com/opencord/voltha-protos/v5/go/voltha"
)
func TestMkUniPortNum(t *testing.T) {
diff --git a/internal/pkg/core/olt_state_transitions.go b/internal/pkg/core/olt_state_transitions.go
index efb1f29..6c3f77d 100644
--- a/internal/pkg/core/olt_state_transitions.go
+++ b/internal/pkg/core/olt_state_transitions.go
@@ -22,7 +22,7 @@
"reflect"
"runtime"
- "github.com/opencord/voltha-lib-go/v6/pkg/log"
+ "github.com/opencord/voltha-lib-go/v7/pkg/log"
)
// DeviceState OLT Device state
diff --git a/internal/pkg/core/openolt.go b/internal/pkg/core/openolt.go
index db7da95..f84910d 100644
--- a/internal/pkg/core/openolt.go
+++ b/internal/pkg/core/openolt.go
@@ -19,30 +19,29 @@
import (
"context"
+ "errors"
"sync"
"time"
- "github.com/opencord/voltha-lib-go/v6/pkg/adapters/adapterif"
- conf "github.com/opencord/voltha-lib-go/v6/pkg/config"
- "github.com/opencord/voltha-lib-go/v6/pkg/events/eventif"
- "github.com/opencord/voltha-lib-go/v6/pkg/kafka"
- "github.com/opencord/voltha-lib-go/v6/pkg/log"
+ "github.com/golang/protobuf/ptypes/empty"
+ conf "github.com/opencord/voltha-lib-go/v7/pkg/config"
+ "github.com/opencord/voltha-lib-go/v7/pkg/events/eventif"
+ vgrpc "github.com/opencord/voltha-lib-go/v7/pkg/grpc"
+ "github.com/opencord/voltha-lib-go/v7/pkg/log"
"github.com/opencord/voltha-openolt-adapter/internal/pkg/config"
"github.com/opencord/voltha-openolt-adapter/internal/pkg/olterrors"
- "github.com/opencord/voltha-protos/v4/go/extension"
- ic "github.com/opencord/voltha-protos/v4/go/inter_container"
- "github.com/opencord/voltha-protos/v4/go/openflow_13"
- "github.com/opencord/voltha-protos/v4/go/voltha"
+ "github.com/opencord/voltha-protos/v5/go/common"
+ "github.com/opencord/voltha-protos/v5/go/extension"
+ ic "github.com/opencord/voltha-protos/v5/go/inter_container"
+ "github.com/opencord/voltha-protos/v5/go/voltha"
)
//OpenOLT structure holds the OLT information
type OpenOLT struct {
configManager *conf.ConfigManager
deviceHandlers map[string]*DeviceHandler
- coreProxy adapterif.CoreProxy
- adapterProxy adapterif.AdapterProxy
+ coreClient *vgrpc.Client
eventProxy eventif.EventProxy
- kafkaICProxy kafka.InterContainerProxy
config *config.AdapterFlags
numOnus int
KVStoreAddress string
@@ -54,20 +53,19 @@
lockDeviceHandlersMap sync.RWMutex
enableONUStats bool
enableGemStats bool
+ rpcTimeout time.Duration
}
//NewOpenOLT returns a new instance of OpenOLT
-func NewOpenOLT(ctx context.Context, kafkaICProxy kafka.InterContainerProxy,
- coreProxy adapterif.CoreProxy, adapterProxy adapterif.AdapterProxy,
+func NewOpenOLT(ctx context.Context,
+ coreClient *vgrpc.Client,
eventProxy eventif.EventProxy, cfg *config.AdapterFlags, cm *conf.ConfigManager) *OpenOLT {
var openOLT OpenOLT
openOLT.exitChannel = make(chan int, 1)
openOLT.deviceHandlers = make(map[string]*DeviceHandler)
- openOLT.kafkaICProxy = kafkaICProxy
openOLT.config = cfg
openOLT.numOnus = cfg.OnuNumber
- openOLT.coreProxy = coreProxy
- openOLT.adapterProxy = adapterProxy
+ openOLT.coreClient = coreClient
openOLT.eventProxy = eventProxy
openOLT.KVStoreAddress = cfg.KVStoreAddress
openOLT.KVStoreType = cfg.KVStoreType
@@ -78,6 +76,7 @@
openOLT.configManager = cm
openOLT.enableONUStats = cfg.EnableONUStats
openOLT.enableGemStats = cfg.EnableGEMStats
+ openOLT.rpcTimeout = cfg.RPCTimeout
return &openOLT
}
@@ -119,260 +118,169 @@
return nil
}
-// Adopt_device creates a new device handler if not present already and then adopts the device
-func (oo *OpenOLT) Adopt_device(ctx context.Context, device *voltha.Device) error {
+// GetHealthStatus is used as a service readiness validation as a grpc connection
+func (oo *OpenOLT) GetHealthStatus(ctx context.Context, empty *empty.Empty) (*voltha.HealthStatus, error) {
+ return &voltha.HealthStatus{State: voltha.HealthStatus_HEALTHY}, nil
+}
+
+// AdoptDevice creates a new device handler if not present already and then adopts the device
+func (oo *OpenOLT) AdoptDevice(ctx context.Context, device *voltha.Device) (*empty.Empty, error) {
if device == nil {
- return olterrors.NewErrInvalidValue(log.Fields{"device": nil}, nil).Log()
+ return nil, olterrors.NewErrInvalidValue(log.Fields{"device": nil}, nil).Log()
}
logger.Infow(ctx, "adopt-device", log.Fields{"device-id": device.Id})
var handler *DeviceHandler
if handler = oo.getDeviceHandler(device.Id); handler == nil {
- handler := NewDeviceHandler(oo.coreProxy, oo.adapterProxy, oo.eventProxy, device, oo, oo.configManager)
+ handler := NewDeviceHandler(oo.coreClient, oo.eventProxy, device, oo, oo.configManager, oo.config)
oo.addDeviceHandlerToMap(handler)
- go handler.AdoptDevice(ctx, device)
- // Launch the creation of the device topic
- // go oo.createDeviceTopic(device)
+ go handler.AdoptDevice(log.WithSpanFromContext(context.Background(), ctx), device)
}
- return nil
+ return &empty.Empty{}, nil
}
-//Get_ofp_device_info returns OFP information for the given device
-func (oo *OpenOLT) Get_ofp_device_info(ctx context.Context, device *voltha.Device) (*ic.SwitchCapability, error) {
- logger.Infow(ctx, "Get_ofp_device_info", log.Fields{"device-id": device.Id})
+//GetOfpDeviceInfo returns OFP information for the given device
+func (oo *OpenOLT) GetOfpDeviceInfo(ctx context.Context, device *voltha.Device) (*ic.SwitchCapability, error) {
+ logger.Infow(ctx, "get_ofp_device_info", log.Fields{"device-id": device.Id})
if handler := oo.getDeviceHandler(device.Id); handler != nil {
return handler.GetOfpDeviceInfo(device)
}
return nil, olterrors.NewErrNotFound("device-handler", log.Fields{"device-id": device.Id}, nil)
}
-//Process_inter_adapter_message sends messages to a target device (between adapters)
-func (oo *OpenOLT) Process_inter_adapter_message(ctx context.Context, msg *ic.InterAdapterMessage) error {
- logger.Debugw(ctx, "Process_inter_adapter_message", log.Fields{"msgId": msg.Header.Id})
- targetDevice := msg.Header.ProxyDeviceId // Request?
- if targetDevice == "" && msg.Header.ToDeviceId != "" {
- // Typical response
- targetDevice = msg.Header.ToDeviceId
- }
- if handler := oo.getDeviceHandler(targetDevice); handler != nil {
- return handler.ProcessInterAdapterMessage(ctx, msg)
- }
- return olterrors.NewErrNotFound("device-handler", log.Fields{"device-id": targetDevice}, nil)
-}
-
-//Process_tech_profile_instance_request processes tech profile request message from onu adapter
-func (oo *OpenOLT) Process_tech_profile_instance_request(ctx context.Context, msg *ic.InterAdapterTechProfileInstanceRequestMessage) *ic.InterAdapterTechProfileDownloadMessage {
- logger.Debugw(ctx, "Process_tech_profile_instance_request", log.Fields{"tpPath": msg.TpInstancePath})
- targetDeviceID := msg.ParentDeviceId // Request?
- if targetDeviceID == "" {
- logger.Error(ctx, "device-id-nil")
- return nil
- }
- if handler := oo.getDeviceHandler(targetDeviceID); handler != nil {
- return handler.GetInterAdapterTechProfileDownloadMessage(ctx, msg.TpInstancePath, msg.ParentPonPort, msg.OnuId, msg.UniId)
- }
- return nil
-}
-
-//Adapter_descriptor not implemented
-func (oo *OpenOLT) Adapter_descriptor(ctx context.Context) error {
- return olterrors.ErrNotImplemented
-}
-
-//Device_types unimplemented
-func (oo *OpenOLT) Device_types(ctx context.Context) (*voltha.DeviceTypes, error) {
- return nil, olterrors.ErrNotImplemented
-}
-
-//Health returns unimplemented
-func (oo *OpenOLT) Health(ctx context.Context) (*voltha.HealthStatus, error) {
- return nil, olterrors.ErrNotImplemented
-}
-
-//Reconcile_device unimplemented
-func (oo *OpenOLT) Reconcile_device(ctx context.Context, device *voltha.Device) error {
+//ReconcileDevice unimplemented
+func (oo *OpenOLT) ReconcileDevice(ctx context.Context, device *voltha.Device) (*empty.Empty, error) {
if device == nil {
- return olterrors.NewErrInvalidValue(log.Fields{"device": nil}, nil)
+ return nil, olterrors.NewErrInvalidValue(log.Fields{"device": nil}, nil)
}
logger.Infow(ctx, "reconcile-device", log.Fields{"device-id": device.Id})
var handler *DeviceHandler
if handler = oo.getDeviceHandler(device.Id); handler == nil {
- handler := NewDeviceHandler(oo.coreProxy, oo.adapterProxy, oo.eventProxy, device, oo, oo.configManager)
+ handler := NewDeviceHandler(oo.coreClient, oo.eventProxy, device, oo, oo.configManager, oo.config)
handler.adapterPreviouslyConnected = true
oo.addDeviceHandlerToMap(handler)
handler.transitionMap = NewTransitionMap(handler)
+
//Setting state to RECONCILING
- err := handler.coreProxy.DeviceStateUpdate(ctx, device.Id, device.ConnectStatus, voltha.OperStatus_RECONCILING)
+ cgClient, err := oo.coreClient.GetCoreServiceClient()
if err != nil {
- return err
+ return nil, err
}
- handler.transitionMap.Handle(ctx, DeviceInit)
+ subCtx, cancel := context.WithTimeout(log.WithSpanFromContext(context.Background(), ctx), oo.rpcTimeout)
+ defer cancel()
+ if _, err := cgClient.DeviceStateUpdate(subCtx, &ic.DeviceStateFilter{
+ DeviceId: device.Id,
+ OperStatus: voltha.OperStatus_RECONCILING,
+ ConnStatus: device.ConnectStatus,
+ }); err != nil {
+ return nil, olterrors.NewErrAdapter("device-update-failed", log.Fields{"device-id": device.Id}, err)
+ }
+
+ handler.transitionMap.Handle(log.WithSpanFromContext(context.Background(), ctx), DeviceInit)
}
- return nil
+ return &empty.Empty{}, nil
}
-//Abandon_device unimplemented
-func (oo *OpenOLT) Abandon_device(ctx context.Context, device *voltha.Device) error {
- return olterrors.ErrNotImplemented
-}
-
-//Disable_device disables the given device
-func (oo *OpenOLT) Disable_device(ctx context.Context, device *voltha.Device) error {
+//DisableDevice disables the given device
+func (oo *OpenOLT) DisableDevice(ctx context.Context, device *voltha.Device) (*empty.Empty, error) {
logger.Infow(ctx, "disable-device", log.Fields{"device-id": device.Id})
if handler := oo.getDeviceHandler(device.Id); handler != nil {
- return handler.DisableDevice(ctx, device)
+ if err := handler.DisableDevice(log.WithSpanFromContext(context.Background(), ctx), device); err != nil {
+ return nil, err
+ }
+ return &empty.Empty{}, nil
}
- return olterrors.NewErrNotFound("device-handler", log.Fields{"device-id": device.Id}, nil)
+ return nil, olterrors.NewErrNotFound("device-handler", log.Fields{"device-id": device.Id}, nil)
}
-//Reenable_device enables the olt device after disable
-func (oo *OpenOLT) Reenable_device(ctx context.Context, device *voltha.Device) error {
+//ReEnableDevice enables the olt device after disable
+func (oo *OpenOLT) ReEnableDevice(ctx context.Context, device *voltha.Device) (*empty.Empty, error) {
logger.Infow(ctx, "reenable-device", log.Fields{"device-id": device.Id})
if handler := oo.getDeviceHandler(device.Id); handler != nil {
- return handler.ReenableDevice(ctx, device)
+ if err := handler.ReenableDevice(log.WithSpanFromContext(context.Background(), ctx), device); err != nil {
+ return nil, err
+ }
+ return &empty.Empty{}, nil
}
- return olterrors.NewErrNotFound("device-handler", log.Fields{"device-id": device.Id}, nil)
+ return nil, olterrors.NewErrNotFound("device-handler", log.Fields{"device-id": device.Id}, nil)
}
-//Reboot_device reboots the given device
-func (oo *OpenOLT) Reboot_device(ctx context.Context, device *voltha.Device) error {
+//RebootDevice reboots the given device
+func (oo *OpenOLT) RebootDevice(ctx context.Context, device *voltha.Device) (*empty.Empty, error) {
logger.Infow(ctx, "reboot-device", log.Fields{"device-id": device.Id})
if handler := oo.getDeviceHandler(device.Id); handler != nil {
- return handler.RebootDevice(ctx, device)
+ if err := handler.RebootDevice(log.WithSpanFromContext(context.Background(), ctx), device); err != nil {
+ return nil, err
+ }
+ return &empty.Empty{}, nil
}
- return olterrors.NewErrNotFound("device-handler", log.Fields{"device-id": device.Id}, nil)
+ return nil, olterrors.NewErrNotFound("device-handler", log.Fields{"device-id": device.Id}, nil)
+
}
-//Self_test_device unimplented
-func (oo *OpenOLT) Self_test_device(ctx context.Context, device *voltha.Device) error {
- return olterrors.ErrNotImplemented
-}
-
-//Delete_device unimplemented
-func (oo *OpenOLT) Delete_device(ctx context.Context, device *voltha.Device) error {
+//DeleteDevice deletes a device
+func (oo *OpenOLT) DeleteDevice(ctx context.Context, device *voltha.Device) (*empty.Empty, error) {
logger.Infow(ctx, "delete-device", log.Fields{"device-id": device.Id})
if handler := oo.getDeviceHandler(device.Id); handler != nil {
- if err := handler.DeleteDevice(ctx, device); err != nil {
+ if err := handler.DeleteDevice(log.WithSpanFromContext(context.Background(), ctx), device); err != nil {
logger.Errorw(ctx, "failed-to-handle-delete-device", log.Fields{"device-id": device.Id})
}
oo.deleteDeviceHandlerToMap(handler)
- return nil
+ return &empty.Empty{}, nil
}
- return olterrors.NewErrNotFound("device-handler", log.Fields{"device-id": device.Id}, nil)
+ return nil, olterrors.NewErrNotFound("device-handler", log.Fields{"device-id": device.Id}, nil)
}
-//Get_device_details unimplemented
-func (oo *OpenOLT) Get_device_details(ctx context.Context, device *voltha.Device) error {
- return olterrors.ErrNotImplemented
-}
-
-//Update_flows_bulk returns
-func (oo *OpenOLT) Update_flows_bulk(ctx context.Context, device *voltha.Device, flows *voltha.Flows, groups *voltha.FlowGroups, flowMetadata *voltha.FlowMetadata) error {
- return olterrors.ErrNotImplemented
-}
-
-//Update_flows_incrementally updates (add/remove) the flows on a given device
-func (oo *OpenOLT) Update_flows_incrementally(ctx context.Context, device *voltha.Device, flows *openflow_13.FlowChanges, groups *openflow_13.FlowGroupChanges, flowMetadata *voltha.FlowMetadata) error {
- logger.Infow(ctx, "Update_flows_incrementally", log.Fields{"device-id": device.Id, "flows": flows, "flowMetadata": flowMetadata})
- if handler := oo.getDeviceHandler(device.Id); handler != nil {
- return handler.UpdateFlowsIncrementally(ctx, device, flows, groups, flowMetadata)
+//UpdateFlowsIncrementally updates (add/remove) the flows on a given device
+func (oo *OpenOLT) UpdateFlowsIncrementally(ctx context.Context, incrFlows *ic.IncrementalFlows) (*empty.Empty, error) {
+ logger.Infow(ctx, "update_flows_incrementally", log.Fields{"device-id": incrFlows.Device.Id, "flows": incrFlows.Flows, "flowMetadata": incrFlows.FlowMetadata})
+ if handler := oo.getDeviceHandler(incrFlows.Device.Id); handler != nil {
+ if err := handler.UpdateFlowsIncrementally(log.WithSpanFromContext(context.Background(), ctx), incrFlows.Device, incrFlows.Flows, incrFlows.Groups, incrFlows.FlowMetadata); err != nil {
+ return nil, err
+ }
+ return &empty.Empty{}, nil
}
- return olterrors.NewErrNotFound("device-handler", log.Fields{"device-id": device.Id}, nil)
+ return nil, olterrors.NewErrNotFound("device-handler", log.Fields{"device-id": incrFlows.Device.Id}, nil)
}
-//Update_pm_config returns PmConfigs nil or error
-func (oo *OpenOLT) Update_pm_config(ctx context.Context, device *voltha.Device, pmConfigs *voltha.PmConfigs) error {
- logger.Debugw(ctx, "Update_pm_config", log.Fields{"device-id": device.Id, "pm-configs": pmConfigs})
- if handler := oo.getDeviceHandler(device.Id); handler != nil {
- handler.UpdatePmConfig(ctx, pmConfigs)
- return nil
+//UpdatePmConfig returns PmConfigs nil or error
+func (oo *OpenOLT) UpdatePmConfig(ctx context.Context, configs *ic.PmConfigsInfo) (*empty.Empty, error) {
+ logger.Debugw(ctx, "update_pm_config", log.Fields{"device-id": configs.DeviceId, "pm-configs": configs.PmConfigs})
+ if handler := oo.getDeviceHandler(configs.DeviceId); handler != nil {
+ handler.UpdatePmConfig(log.WithSpanFromContext(context.Background(), ctx), configs.PmConfigs)
+ return &empty.Empty{}, nil
}
- return olterrors.NewErrNotFound("device-handler", log.Fields{"device-id": device.Id}, nil)
+ return nil, olterrors.NewErrNotFound("device-handler", log.Fields{"device-id": configs.DeviceId}, nil)
}
-//Receive_packet_out sends packet out to the device
-func (oo *OpenOLT) Receive_packet_out(ctx context.Context, deviceID string, egressPortNo int, packet *openflow_13.OfpPacketOut) error {
- logger.Debugw(ctx, "Receive_packet_out", log.Fields{"device-id": deviceID, "egress_port_no": egressPortNo, "pkt": packet})
- if handler := oo.getDeviceHandler(deviceID); handler != nil {
- return handler.PacketOut(ctx, egressPortNo, packet)
+//SendPacketOut sends packet out to the device
+func (oo *OpenOLT) SendPacketOut(ctx context.Context, packet *ic.PacketOut) (*empty.Empty, error) {
+ logger.Debugw(ctx, "send_packet_out", log.Fields{"device-id": packet.DeviceId, "egress_port_no": packet.EgressPortNo, "pkt": packet.Packet})
+ if handler := oo.getDeviceHandler(packet.DeviceId); handler != nil {
+ if err := handler.PacketOut(log.WithSpanFromContext(context.Background(), ctx), packet.EgressPortNo, packet.Packet); err != nil {
+ return nil, err
+ }
+ return &empty.Empty{}, nil
}
- return olterrors.NewErrNotFound("device-handler", log.Fields{"device-id": deviceID}, nil)
+ return nil, olterrors.NewErrNotFound("device-handler", log.Fields{"device-id": packet.DeviceId}, nil)
+
}
-//Suppress_event unimplemented
-func (oo *OpenOLT) Suppress_event(ctx context.Context, filter *voltha.EventFilter) error {
- return olterrors.ErrNotImplemented
+// EnablePort to Enable PON/NNI interface
+func (oo *OpenOLT) EnablePort(ctx context.Context, port *voltha.Port) (*empty.Empty, error) {
+ logger.Infow(ctx, "enable_port", log.Fields{"device-id": port.DeviceId, "port": port})
+ if err := oo.enableDisablePort(log.WithSpanFromContext(context.Background(), ctx), port.DeviceId, port, true); err != nil {
+ return nil, err
+ }
+ return &empty.Empty{}, nil
}
-//Unsuppress_event unimplemented
-func (oo *OpenOLT) Unsuppress_event(ctx context.Context, filter *voltha.EventFilter) error {
- return olterrors.ErrNotImplemented
-}
-
-//Download_image unimplemented
-func (oo *OpenOLT) Download_image(ctx context.Context, device *voltha.Device, request *voltha.ImageDownload) (*voltha.ImageDownload, error) {
- return nil, olterrors.ErrNotImplemented
-}
-
-//Get_image_download_status unimplemented
-func (oo *OpenOLT) Get_image_download_status(ctx context.Context, device *voltha.Device, request *voltha.ImageDownload) (*voltha.ImageDownload, error) {
- return nil, olterrors.ErrNotImplemented
-}
-
-//Cancel_image_download unimplemented
-func (oo *OpenOLT) Cancel_image_download(ctx context.Context, device *voltha.Device, request *voltha.ImageDownload) (*voltha.ImageDownload, error) {
- return nil, olterrors.ErrNotImplemented
-}
-
-//Activate_image_update unimplemented
-func (oo *OpenOLT) Activate_image_update(ctx context.Context, device *voltha.Device, request *voltha.ImageDownload) (*voltha.ImageDownload, error) {
- return nil, olterrors.ErrNotImplemented
-}
-
-//Revert_image_update unimplemented
-func (oo *OpenOLT) Revert_image_update(ctx context.Context, device *voltha.Device, request *voltha.ImageDownload) (*voltha.ImageDownload, error) {
- return nil, olterrors.ErrNotImplemented
-}
-
-//Download_onu_image unimplemented
-func (oo *OpenOLT) Download_onu_image(ctx context.Context, request *voltha.DeviceImageDownloadRequest) (*voltha.DeviceImageResponse, error) {
- return nil, olterrors.ErrNotImplemented
-}
-
-//Get_onu_image_status unimplemented
-func (oo *OpenOLT) Get_onu_image_status(ctx context.Context, in *voltha.DeviceImageRequest) (*voltha.DeviceImageResponse, error) {
- return nil, olterrors.ErrNotImplemented
-}
-
-//Abort_onu_image_upgrade unimplemented
-func (oo *OpenOLT) Abort_onu_image_upgrade(ctx context.Context, in *voltha.DeviceImageRequest) (*voltha.DeviceImageResponse, error) {
- return nil, olterrors.ErrNotImplemented
-}
-
-//Get_onu_images unimplemented
-func (oo *OpenOLT) Get_onu_images(ctx context.Context, deviceID string) (*voltha.OnuImages, error) {
- return nil, olterrors.ErrNotImplemented
-}
-
-//Activate_onu_image unimplemented
-func (oo *OpenOLT) Activate_onu_image(ctx context.Context, in *voltha.DeviceImageRequest) (*voltha.DeviceImageResponse, error) {
- return nil, olterrors.ErrNotImplemented
-}
-
-//Commit_onu_image unimplemented
-func (oo *OpenOLT) Commit_onu_image(ctx context.Context, in *voltha.DeviceImageRequest) (*voltha.DeviceImageResponse, error) {
- return nil, olterrors.ErrNotImplemented
-}
-
-// Enable_port to Enable PON/NNI interface
-func (oo *OpenOLT) Enable_port(ctx context.Context, deviceID string, port *voltha.Port) error {
- logger.Infow(ctx, "Enable_port", log.Fields{"device-id": deviceID, "port": port})
- return oo.enableDisablePort(ctx, deviceID, port, true)
-}
-
-// Disable_port to Disable pon/nni interface
-func (oo *OpenOLT) Disable_port(ctx context.Context, deviceID string, port *voltha.Port) error {
- logger.Infow(ctx, "Disable_port", log.Fields{"device-id": deviceID, "port": port})
- return oo.enableDisablePort(ctx, deviceID, port, false)
+// DisablePort to Disable pon/nni interface
+func (oo *OpenOLT) DisablePort(ctx context.Context, port *voltha.Port) (*empty.Empty, error) {
+ logger.Infow(ctx, "disable_port", log.Fields{"device-id": port.DeviceId, "port": port})
+ if err := oo.enableDisablePort(log.WithSpanFromContext(context.Background(), ctx), port.DeviceId, port, false); err != nil {
+ return nil, err
+ }
+ return &empty.Empty{}, nil
}
// enableDisablePort to Disable pon or Enable PON interface
@@ -399,38 +307,36 @@
return nil
}
-//Child_device_lost deletes the ONU and its references from PONResources
-func (oo *OpenOLT) Child_device_lost(ctx context.Context, childDevice *voltha.Device) error {
+//ChildDeviceLost deletes the ONU and its references from PONResources
+func (oo *OpenOLT) ChildDeviceLost(ctx context.Context, childDevice *voltha.Device) (*empty.Empty, error) {
logger.Infow(ctx, "Child-device-lost", log.Fields{"parent-device-id": childDevice.ParentId, "child-device-id": childDevice.Id})
if handler := oo.getDeviceHandler(childDevice.ParentId); handler != nil {
- return handler.ChildDeviceLost(ctx, childDevice.ParentPortNo, childDevice.ProxyAddress.OnuId, childDevice.SerialNumber)
+ if err := handler.ChildDeviceLost(log.WithSpanFromContext(context.Background(), ctx), childDevice.ParentPortNo, childDevice.ProxyAddress.OnuId, childDevice.SerialNumber); err != nil {
+ return nil, err
+ }
+ return &empty.Empty{}, nil
}
- return olterrors.NewErrNotFound("device-handler", log.Fields{"device-id": childDevice.ParentId}, nil).Log()
+ return nil, olterrors.NewErrNotFound("device-handler", log.Fields{"parent-device-id": childDevice.ParentId}, nil).Log()
}
-//Start_omci_test not implemented
-func (oo *OpenOLT) Start_omci_test(ctx context.Context, device *voltha.Device, request *voltha.OmciTestRequest) (*voltha.TestResponse, error) {
- return nil, olterrors.ErrNotImplemented
-}
-
-//Get_ext_value retrieves a value on a particular ONU
-func (oo *OpenOLT) Get_ext_value(ctx context.Context, deviceID string, device *voltha.Device, valueparam voltha.ValueType_Type) (*voltha.ReturnValues, error) {
+// GetExtValue retrieves a value on a particular ONU
+func (oo *OpenOLT) GetExtValue(ctx context.Context, extInfo *ic.GetExtValueMessage) (*voltha.ReturnValues, error) {
var err error
resp := new(voltha.ReturnValues)
- logger.Infow(ctx, "Get_ext_value", log.Fields{"device-id": deviceID, "onu-id": device.Id})
- if handler := oo.getDeviceHandler(deviceID); handler != nil {
- if resp, err = handler.getExtValue(ctx, device, valueparam); err != nil {
- logger.Errorw(ctx, "error-occurred-during-get-ext-value", log.Fields{"device-id": deviceID, "onu-id": device.Id,
- "err": err})
+ logger.Infow(ctx, "get_ext_value", log.Fields{"parent-device-id": extInfo.ParentDevice.Id, "onu-id": extInfo.ChildDevice.Id})
+ if handler := oo.getDeviceHandler(extInfo.ParentDevice.Id); handler != nil {
+ if resp, err = handler.getExtValue(ctx, extInfo.ChildDevice, extInfo.ValueType); err != nil {
+ logger.Errorw(ctx, "error-occurred-during-get-ext-value",
+ log.Fields{"parent-device-id": extInfo.ParentDevice.Id, "onu-id": extInfo.ChildDevice.Id, "error": err})
return nil, err
}
}
return resp, nil
}
-//Single_get_value_request handles get uni status on ONU and ondemand metric on OLT
-func (oo *OpenOLT) Single_get_value_request(ctx context.Context, request extension.SingleGetValueRequest) (*extension.SingleGetValueResponse, error) {
- logger.Infow(ctx, "Single_get_value_request", log.Fields{"request": request})
+//GetSingleValue handles get uni status on ONU and ondemand metric on OLT
+func (oo *OpenOLT) GetSingleValue(ctx context.Context, request *extension.SingleGetValueRequest) (*extension.SingleGetValueResponse, error) {
+ logger.Infow(ctx, "single_get_value_request", log.Fields{"request": request})
errResp := func(status extension.GetValueResponse_Status,
reason extension.GetValueResponse_ErrorReason) *extension.SingleGetValueResponse {
@@ -457,3 +363,136 @@
logger.Infow(ctx, "Single_get_value_request failed ", log.Fields{"request": request})
return errResp(extension.GetValueResponse_ERROR, extension.GetValueResponse_INVALID_DEVICE_ID), nil
}
+
+/*
+ * OLT Inter-adapter service
+ */
+
+// ProxyOmciRequest proxies an OMCI request from the child adapter
+func (oo *OpenOLT) ProxyOmciRequest(ctx context.Context, request *ic.OmciMessage) (*empty.Empty, error) {
+ logger.Debugw(ctx, "proxy-omci-request", log.Fields{"request": request})
+
+ if handler := oo.getDeviceHandler(request.ParentDeviceId); handler != nil {
+ if err := handler.ProxyOmciMessage(ctx, request); err != nil {
+ return nil, errors.New(err.Error())
+ }
+ return &empty.Empty{}, nil
+ }
+ return nil, olterrors.NewErrNotFound("no-device-handler", log.Fields{"parent-device-id": request.ParentDeviceId, "child-device-id": request.ChildDeviceId}, nil).Log()
+}
+
+// GetTechProfileInstance returns an instance of a tech profile
+func (oo *OpenOLT) GetTechProfileInstance(ctx context.Context, request *ic.TechProfileInstanceRequestMessage) (*ic.TechProfileDownloadMessage, error) {
+ logger.Debugw(ctx, "getting-tech-profile-request", log.Fields{"request": request})
+
+ targetDeviceID := request.ParentDeviceId
+ if targetDeviceID == "" {
+ return nil, olterrors.NewErrNotFound("parent-id-empty", log.Fields{"parent-device-id": request.ParentDeviceId, "child-device-id": request.DeviceId}, nil).Log()
+ }
+ if handler := oo.getDeviceHandler(targetDeviceID); handler != nil {
+ return handler.GetTechProfileDownloadMessage(ctx, request)
+ }
+ return nil, olterrors.NewErrNotFound("no-device-handler", log.Fields{"parent-device-id": request.ParentDeviceId, "child-device-id": request.DeviceId}, nil).Log()
+
+}
+
+/*
+ *
+ * Unimplemented APIs
+ *
+ */
+
+//SimulateAlarm is unimplemented
+func (oo *OpenOLT) SimulateAlarm(context.Context, *ic.SimulateAlarmMessage) (*voltha.OperationResp, error) {
+ return nil, olterrors.ErrNotImplemented
+}
+
+//SetExtValue is unimplemented
+func (oo *OpenOLT) SetExtValue(context.Context, *ic.SetExtValueMessage) (*empty.Empty, error) {
+ return nil, olterrors.ErrNotImplemented
+}
+
+//SetSingleValue is unimplemented
+func (oo *OpenOLT) SetSingleValue(context.Context, *extension.SingleSetValueRequest) (*extension.SingleSetValueResponse, error) {
+ return nil, olterrors.ErrNotImplemented
+}
+
+//StartOmciTest not implemented
+func (oo *OpenOLT) StartOmciTest(ctx context.Context, test *ic.OMCITest) (*voltha.TestResponse, error) {
+ return nil, olterrors.ErrNotImplemented
+}
+
+//SuppressEvent unimplemented
+func (oo *OpenOLT) SuppressEvent(ctx context.Context, filter *voltha.EventFilter) (*empty.Empty, error) {
+ return nil, olterrors.ErrNotImplemented
+}
+
+//UnSuppressEvent unimplemented
+func (oo *OpenOLT) UnSuppressEvent(ctx context.Context, filter *voltha.EventFilter) (*empty.Empty, error) {
+ return nil, olterrors.ErrNotImplemented
+}
+
+//DownloadImage is unimplemented
+func (oo *OpenOLT) DownloadImage(ctx context.Context, imageInfo *ic.ImageDownloadMessage) (*voltha.ImageDownload, error) {
+ return nil, olterrors.ErrNotImplemented
+}
+
+//GetImageDownloadStatus is unimplemented
+func (oo *OpenOLT) GetImageDownloadStatus(ctx context.Context, imageInfo *ic.ImageDownloadMessage) (*voltha.ImageDownload, error) {
+ return nil, olterrors.ErrNotImplemented
+}
+
+//CancelImageDownload is unimplemented
+func (oo *OpenOLT) CancelImageDownload(ctx context.Context, imageInfo *ic.ImageDownloadMessage) (*voltha.ImageDownload, error) {
+ return nil, olterrors.ErrNotImplemented
+}
+
+//ActivateImageUpdate is unimplemented
+func (oo *OpenOLT) ActivateImageUpdate(ctx context.Context, imageInfo *ic.ImageDownloadMessage) (*voltha.ImageDownload, error) {
+ return nil, olterrors.ErrNotImplemented
+}
+
+//RevertImageUpdate is unimplemented
+func (oo *OpenOLT) RevertImageUpdate(ctx context.Context, imageInfo *ic.ImageDownloadMessage) (*voltha.ImageDownload, error) {
+ return nil, olterrors.ErrNotImplemented
+}
+
+//DownloadOnuImage unimplemented
+func (oo *OpenOLT) DownloadOnuImage(ctx context.Context, request *voltha.DeviceImageDownloadRequest) (*voltha.DeviceImageResponse, error) {
+ return nil, olterrors.ErrNotImplemented
+}
+
+//GetOnuImageStatus unimplemented
+func (oo *OpenOLT) GetOnuImageStatus(ctx context.Context, in *voltha.DeviceImageRequest) (*voltha.DeviceImageResponse, error) {
+ return nil, olterrors.ErrNotImplemented
+}
+
+//AbortOnuImageUpgrade unimplemented
+func (oo *OpenOLT) AbortOnuImageUpgrade(ctx context.Context, in *voltha.DeviceImageRequest) (*voltha.DeviceImageResponse, error) {
+ return nil, olterrors.ErrNotImplemented
+}
+
+//GetOnuImages unimplemented
+func (oo *OpenOLT) GetOnuImages(ctx context.Context, deviceID *common.ID) (*voltha.OnuImages, error) {
+ return nil, olterrors.ErrNotImplemented
+}
+
+//ActivateOnuImage unimplemented
+func (oo *OpenOLT) ActivateOnuImage(ctx context.Context, in *voltha.DeviceImageRequest) (*voltha.DeviceImageResponse, error) {
+ return nil, olterrors.ErrNotImplemented
+}
+
+//CommitOnuImage unimplemented
+func (oo *OpenOLT) CommitOnuImage(ctx context.Context, in *voltha.DeviceImageRequest) (*voltha.DeviceImageResponse, error) {
+ return nil, olterrors.ErrNotImplemented
+}
+
+// UpdateFlowsBulk is unimplemented
+func (oo *OpenOLT) UpdateFlowsBulk(ctx context.Context, flows *ic.BulkFlows) (*empty.Empty, error) {
+ return nil, olterrors.ErrNotImplemented
+}
+
+//SelfTestDevice unimplemented
+func (oo *OpenOLT) SelfTestDevice(ctx context.Context, device *voltha.Device) (*empty.Empty, error) {
+ return nil, olterrors.ErrNotImplemented
+}
diff --git a/internal/pkg/core/openolt_eventmgr.go b/internal/pkg/core/openolt_eventmgr.go
index e72668e..45b327b 100644
--- a/internal/pkg/core/openolt_eventmgr.go
+++ b/internal/pkg/core/openolt_eventmgr.go
@@ -23,12 +23,14 @@
"fmt"
"strconv"
- "github.com/opencord/voltha-lib-go/v6/pkg/events/eventif"
- "github.com/opencord/voltha-lib-go/v6/pkg/log"
+ ic "github.com/opencord/voltha-protos/v5/go/inter_container"
+
+ "github.com/opencord/voltha-lib-go/v7/pkg/events/eventif"
+ "github.com/opencord/voltha-lib-go/v7/pkg/log"
"github.com/opencord/voltha-openolt-adapter/internal/pkg/olterrors"
- "github.com/opencord/voltha-protos/v4/go/common"
- oop "github.com/opencord/voltha-protos/v4/go/openolt"
- "github.com/opencord/voltha-protos/v4/go/voltha"
+ "github.com/opencord/voltha-protos/v5/go/common"
+ oop "github.com/opencord/voltha-protos/v5/go/openolt"
+ "github.com/opencord/voltha-protos/v5/go/voltha"
)
const (
@@ -463,7 +465,7 @@
/* Update onu device with LoS raised state as true */
em.handler.onus.Store(onuKey, NewOnuDevice(onuInCache.(*OnuDevice).deviceID, onuInCache.(*OnuDevice).deviceType,
onuInCache.(*OnuDevice).serialNumber, onuInCache.(*OnuDevice).onuID, onuInCache.(*OnuDevice).intfID,
- onuInCache.(*OnuDevice).proxyDeviceID, true))
+ onuInCache.(*OnuDevice).proxyDeviceID, true, onuInCache.(*OnuDevice).adapterEndpoint))
}
case statusCheckOff:
if em.wasLosCleared(ctx, onuAlarm) {
@@ -476,7 +478,7 @@
/* Update onu device with LoS raised state as false */
em.handler.onus.Store(onuKey, NewOnuDevice(onuInCache.(*OnuDevice).deviceID, onuInCache.(*OnuDevice).deviceType,
onuInCache.(*OnuDevice).serialNumber, onuInCache.(*OnuDevice).onuID, onuInCache.(*OnuDevice).intfID,
- onuInCache.(*OnuDevice).proxyDeviceID, false))
+ onuInCache.(*OnuDevice).proxyDeviceID, false, onuInCache.(*OnuDevice).adapterEndpoint))
}
}
@@ -666,7 +668,10 @@
// oltIntfOperIndication handles Up and Down state of an OLT PON ports
func (em *OpenOltEventMgr) oltIntfOperIndication(ctx context.Context, ifindication *oop.IntfOperIndication, deviceID string, raisedTs int64) {
portNo := IntfIDToPortNo(ifindication.IntfId, voltha.Port_PON_OLT)
- if port, err := em.handler.coreProxy.GetDevicePort(ctx, deviceID, portNo); err != nil {
+ if port, err := em.handler.getPortFromCore(ctx, &ic.PortFilter{
+ DeviceId: deviceID,
+ Port: portNo,
+ }); err != nil {
logger.Warnw(ctx, "Error while fetching port object", log.Fields{"device-id": deviceID, "err": err})
} else if port.AdminState != common.AdminState_ENABLED {
logger.Debugw(ctx, "port-disable/enable-event-not-generated--the-port-is-not-enabled-by-operator", log.Fields{"device-id": deviceID, "port": port})
diff --git a/internal/pkg/core/openolt_eventmgr_test.go b/internal/pkg/core/openolt_eventmgr_test.go
index 955bee9..63361ce 100644
--- a/internal/pkg/core/openolt_eventmgr_test.go
+++ b/internal/pkg/core/openolt_eventmgr_test.go
@@ -24,7 +24,7 @@
"time"
"github.com/opencord/voltha-openolt-adapter/pkg/mocks"
- oop "github.com/opencord/voltha-protos/v4/go/openolt"
+ oop "github.com/opencord/voltha-protos/v5/go/openolt"
)
func mockEventMgr() *OpenOltEventMgr {
@@ -35,8 +35,8 @@
deviceType: "ONU",
serialNumber: "TEST_ONU_123",
onuID: 1, intfID: 1})
- dh.onus.Store("1.3", NewOnuDevice("onu3", "onu3", "onu3", 1, 3, "onu3", false))
- dh.onus.Store("1.4", NewOnuDevice("onu4", "onu4", "onu4", 1, 4, "onu4", false))
+ dh.onus.Store("1.3", NewOnuDevice("onu3", "onu3", "onu3", 1, 3, "onu3", false, "mock-endpoint"))
+ dh.onus.Store("1.4", NewOnuDevice("onu4", "onu4", "onu4", 1, 4, "onu4", false, "mock-endpoint"))
return NewEventMgr(ep, dh)
}
func TestOpenOltEventMgr_ProcessEvents(t *testing.T) {
diff --git a/internal/pkg/core/openolt_flowmgr.go b/internal/pkg/core/openolt_flowmgr.go
index b79a591..bd0efd2 100644
--- a/internal/pkg/core/openolt_flowmgr.go
+++ b/internal/pkg/core/openolt_flowmgr.go
@@ -27,18 +27,18 @@
"sync"
"time"
- "github.com/opencord/voltha-lib-go/v6/pkg/meters"
+ "github.com/opencord/voltha-lib-go/v7/pkg/meters"
- "github.com/opencord/voltha-lib-go/v6/pkg/flows"
- "github.com/opencord/voltha-lib-go/v6/pkg/log"
- tp "github.com/opencord/voltha-lib-go/v6/pkg/techprofile"
+ "github.com/opencord/voltha-lib-go/v7/pkg/flows"
+ "github.com/opencord/voltha-lib-go/v7/pkg/log"
+ tp "github.com/opencord/voltha-lib-go/v7/pkg/techprofile"
rsrcMgr "github.com/opencord/voltha-openolt-adapter/internal/pkg/resourcemanager"
- "github.com/opencord/voltha-protos/v4/go/common"
- ic "github.com/opencord/voltha-protos/v4/go/inter_container"
- ofp "github.com/opencord/voltha-protos/v4/go/openflow_13"
- openoltpb2 "github.com/opencord/voltha-protos/v4/go/openolt"
- tp_pb "github.com/opencord/voltha-protos/v4/go/tech_profile"
- "github.com/opencord/voltha-protos/v4/go/voltha"
+ "github.com/opencord/voltha-protos/v5/go/common"
+ ic "github.com/opencord/voltha-protos/v5/go/inter_container"
+ ofp "github.com/opencord/voltha-protos/v5/go/openflow_13"
+ openoltpb2 "github.com/opencord/voltha-protos/v5/go/openolt"
+ tp_pb "github.com/opencord/voltha-protos/v5/go/tech_profile"
+ "github.com/opencord/voltha-protos/v5/go/voltha"
"github.com/opencord/voltha-openolt-adapter/internal/pkg/olterrors"
"google.golang.org/grpc/codes"
@@ -1719,7 +1719,7 @@
"intf-id": intfID,
"device-id": f.deviceHandler.device.Id}, err)
}
- onuDev = NewOnuDevice(onuDevice.Id, onuDevice.Type, onuDevice.SerialNumber, onuDevice.ProxyAddress.OnuId, onuDevice.ProxyAddress.ChannelId, onuDevice.ProxyAddress.DeviceId, false)
+ onuDev = NewOnuDevice(onuDevice.Id, onuDevice.Type, onuDevice.SerialNumber, onuDevice.ProxyAddress.OnuId, onuDevice.ProxyAddress.ChannelId, onuDevice.ProxyAddress.DeviceId, false, onuDevice.AdapterEndpoint)
//better to ad the device to cache here.
f.deviceHandler.StoreOnuDevice(onuDev.(*OnuDevice))
} else {
@@ -1770,32 +1770,31 @@
return err
}
- delGemPortMsg := &ic.InterAdapterDeleteGemPortMessage{UniId: uniID, TpInstancePath: tpPath, GemPortId: gemPortID}
- logger.Infow(ctx, "sending-gem-port-delete-to-openonu-adapter",
- log.Fields{
- "msg": *delGemPortMsg,
- "device-id": f.deviceHandler.device.Id})
- if sendErr := f.deviceHandler.AdapterProxy.SendInterAdapterMessage(log.WithSpanFromContext(context.Background(), ctx),
- delGemPortMsg,
- ic.InterAdapterMessageType_DELETE_GEM_PORT_REQUEST,
- f.deviceHandler.openOLT.config.Topic,
- onuDev.deviceType,
- onuDev.deviceID,
- onuDev.proxyDeviceID, ""); sendErr != nil {
+ delGemPortMsg := &ic.DeleteGemPortMessage{
+ DeviceId: onuDev.deviceID,
+ UniId: uniID,
+ TpInstancePath: tpPath,
+ GemPortId: gemPortID,
+ }
+ logger.Debugw(ctx, "sending-gem-port-delete-to-openonu-adapter", log.Fields{"msg": *delGemPortMsg, "child-device-id": onuDev.deviceID})
+
+ if err := f.deviceHandler.sendDeleteGemPortToChildAdapter(ctx, onuDev.adapterEndpoint, delGemPortMsg); err != nil {
return olterrors.NewErrCommunication("send-delete-gem-port-to-onu-adapter",
log.Fields{
- "from-adapter": f.deviceHandler.openOLT.config.Topic,
- "to-adapter": onuDev.deviceType,
+ "from-adapter": f.deviceHandler.openOLT.config.AdapterEndpoint,
+ "to-adapter": onuDev.adapterEndpoint,
"onu-id": onuDev.deviceID,
"proxyDeviceID": onuDev.proxyDeviceID,
- "device-id": f.deviceHandler.device.Id}, sendErr)
+ "device-id": f.deviceHandler.device.Id}, err)
}
+
logger.Infow(ctx, "success-sending-del-gem-port-to-onu-adapter",
log.Fields{
- "msg": delGemPortMsg,
- "from-adapter": f.deviceHandler.device.Type,
- "to-adapter": onuDev.deviceType,
- "device-id": f.deviceHandler.device.Id})
+ "msg": delGemPortMsg,
+ "from-adapter": f.deviceHandler.device.Type,
+ "to-adapter": onuDev.deviceType,
+ "device-id": f.deviceHandler.device.Id,
+ "child-device-id": onuDev.deviceID})
return nil
}
@@ -1811,29 +1810,33 @@
return err
}
- delTcontMsg := &ic.InterAdapterDeleteTcontMessage{UniId: uniID, TpInstancePath: tpPath, AllocId: allocID}
+ delTcontMsg := &ic.DeleteTcontMessage{
+ DeviceId: onuDev.deviceID,
+ UniId: uniID,
+ TpInstancePath: tpPath,
+ AllocId: allocID,
+ }
+
logger.Debugw(ctx, "sending-tcont-delete-to-openonu-adapter",
log.Fields{
"msg": *delTcontMsg,
"device-id": f.deviceHandler.device.Id})
- if sendErr := f.deviceHandler.AdapterProxy.SendInterAdapterMessage(log.WithSpanFromContext(context.Background(), ctx),
- delTcontMsg,
- ic.InterAdapterMessageType_DELETE_TCONT_REQUEST,
- f.deviceHandler.openOLT.config.Topic,
- onuDev.deviceType,
- onuDev.deviceID,
- onuDev.proxyDeviceID, ""); sendErr != nil {
+
+ if err := f.deviceHandler.sendDeleteTContToChildAdapter(ctx, onuDev.adapterEndpoint, delTcontMsg); err != nil {
return olterrors.NewErrCommunication("send-delete-tcont-to-onu-adapter",
log.Fields{
- "from-adapter": f.deviceHandler.openOLT.config.Topic,
- "to-adapter": onuDev.deviceType, "onu-id": onuDev.deviceID,
+ "from-adapter": f.deviceHandler.openOLT.config.AdapterEndpoint,
+ "to-adapter": onuDev.adapterEndpoint,
+ "onu-id": onuDev.deviceID,
"proxyDeviceID": onuDev.proxyDeviceID,
- "device-id": f.deviceHandler.device.Id}, sendErr)
+ "device-id": f.deviceHandler.device.Id}, err)
+
}
logger.Infow(ctx, "success-sending-del-tcont-to-onu-adapter",
log.Fields{
- "msg": delTcontMsg,
- "device-id": f.deviceHandler.device.Id})
+ "msg": delTcontMsg,
+ "device-id": f.deviceHandler.device.Id,
+ "child-device-id": onuDev.deviceID})
return nil
}
@@ -1977,7 +1980,8 @@
"onu-id": onuID,
"uni-id": uniID,
"device-id": f.deviceHandler.device.Id,
- "alloc-id": techprofileInst.AllocId})
+ "alloc-id": techprofileInst.AllocId,
+ "error": err})
}
default:
logger.Errorw(ctx, "error-unknown-tech",
@@ -2178,7 +2182,7 @@
f.incomingFlows[onuID] <- flowCb
// Wait on the channel for flow handlers return value
err := <-errChan
- logger.Infow(ctx, "process-flow--received-resp", log.Fields{"err": err, "totalTimeSeconds": time.Since(startTime).Seconds()})
+ logger.Infow(ctx, "process-flow-received-resp", log.Fields{"err": err, "totalTimeSeconds": time.Since(startTime).Seconds()})
return err
}
logger.Errorw(ctx, "flow handler routine not active for onu", log.Fields{"onuID": onuID, "ponPortIdx": f.ponPortIdx})
@@ -2420,26 +2424,22 @@
logger.Debugw(ctx, "got-child-device-from-olt-device-handler", log.Fields{"onu-id": onuDev.deviceID})
tpPath := f.getTPpath(ctx, intfID, uni, TpID)
- tpDownloadMsg := &ic.InterAdapterTechProfileDownloadMessage{
+ tpDownloadMsg := &ic.TechProfileDownloadMessage{
+ DeviceId: onuDev.deviceID,
UniId: uniID,
TpInstancePath: tpPath,
- TechTpInstance: &ic.InterAdapterTechProfileDownloadMessage_TpInstance{TpInstance: &tpInst},
+ TechTpInstance: &ic.TechProfileDownloadMessage_TpInstance{TpInstance: &tpInst},
}
logger.Debugw(ctx, "sending-load-tech-profile-request-to-brcm-onu-adapter", log.Fields{"tpDownloadMsg": *tpDownloadMsg})
- sendErr := f.deviceHandler.AdapterProxy.SendInterAdapterMessage(log.WithSpanFromContext(context.Background(), ctx),
- tpDownloadMsg,
- ic.InterAdapterMessageType_TECH_PROFILE_DOWNLOAD_REQUEST,
- f.deviceHandler.openOLT.config.Topic,
- onuDev.deviceType,
- onuDev.deviceID,
- onuDev.proxyDeviceID, "")
- if sendErr != nil {
+
+ err = f.deviceHandler.sendDownloadTechProfileToChildAdapter(ctx, onuDev.adapterEndpoint, tpDownloadMsg)
+ if err != nil {
return olterrors.NewErrCommunication("send-techprofile-download-request",
log.Fields{
- "from-adapter": f.deviceHandler.openOLT.config.Topic,
+ "from-adapter": f.deviceHandler.openOLT.config.AdapterEndpoint,
"to-adapter": onuDev.deviceType,
"onu-id": onuDev.deviceID,
- "proxyDeviceID": onuDev.proxyDeviceID}, sendErr)
+ "proxyDeviceID": onuDev.proxyDeviceID}, err)
}
logger.Infow(ctx, "success-sending-load-tech-profile-request-to-brcm-onu-adapter", log.Fields{"tpDownloadMsg": *tpDownloadMsg})
return nil
@@ -3351,29 +3351,38 @@
return nil
}
-func (f *OpenOltFlowMgr) getTechProfileDownloadMessage(ctx context.Context, tpPath string, ponID uint32, onuID uint32, uniID uint32) *ic.InterAdapterTechProfileDownloadMessage {
+func (f *OpenOltFlowMgr) getTechProfileDownloadMessage(ctx context.Context, tpPath string, uniID uint32, onuDeviceID string) (*ic.TechProfileDownloadMessage, error) {
tpInst, err := f.techprofile.GetTPInstance(ctx, tpPath)
if err != nil {
logger.Errorw(ctx, "error-fetching-tp-instance", log.Fields{"tpPath": tpPath})
- return nil
+ return nil, err
}
switch tpInst := tpInst.(type) {
case *tp_pb.TechProfileInstance:
- logger.Debugw(ctx, "fetched-tp-instance-successfully--formulating-tp-download-msg", log.Fields{"tpPath": tpPath})
- return &ic.InterAdapterTechProfileDownloadMessage{UniId: uniID,
+ logger.Debugw(ctx, "fetched-tp-instance-successfully-formulating-tp-download-msg", log.Fields{"tpPath": tpPath})
+ return &ic.TechProfileDownloadMessage{
+ DeviceId: onuDeviceID,
+ UniId: uniID,
TpInstancePath: tpPath,
- TechTpInstance: &ic.InterAdapterTechProfileDownloadMessage_TpInstance{TpInstance: tpInst},
- }
+ TechTpInstance: &ic.TechProfileDownloadMessage_TpInstance{TpInstance: tpInst},
+ }, nil
case *openoltpb2.EponTechProfileInstance:
- return &ic.InterAdapterTechProfileDownloadMessage{UniId: uniID,
+ return &ic.TechProfileDownloadMessage{
+ DeviceId: onuDeviceID,
+ UniId: uniID,
TpInstancePath: tpPath,
- TechTpInstance: &ic.InterAdapterTechProfileDownloadMessage_EponTpInstance{EponTpInstance: tpInst},
- }
+ TechTpInstance: &ic.TechProfileDownloadMessage_EponTpInstance{EponTpInstance: tpInst},
+ }, nil
default:
logger.Errorw(ctx, "unknown-tech", log.Fields{"tpPath": tpPath})
}
- return nil
+ return &ic.TechProfileDownloadMessage{
+ DeviceId: onuDeviceID,
+ UniId: uniID,
+ TpInstancePath: tpPath,
+ TechTpInstance: nil,
+ }, nil
}
func (f *OpenOltFlowMgr) getOnuGemInfoList(ctx context.Context) []rsrcMgr.OnuGemInfo {
diff --git a/internal/pkg/core/openolt_flowmgr_test.go b/internal/pkg/core/openolt_flowmgr_test.go
index b985b4e..d636a73 100644
--- a/internal/pkg/core/openolt_flowmgr_test.go
+++ b/internal/pkg/core/openolt_flowmgr_test.go
@@ -21,21 +21,22 @@
"context"
"encoding/hex"
"fmt"
- "github.com/opencord/voltha-openolt-adapter/pkg/mocks"
"reflect"
"strconv"
"sync"
"testing"
"time"
- "github.com/opencord/voltha-protos/v4/go/voltha"
+ "github.com/opencord/voltha-openolt-adapter/pkg/mocks"
- fu "github.com/opencord/voltha-lib-go/v6/pkg/flows"
- "github.com/opencord/voltha-lib-go/v6/pkg/log"
+ "github.com/opencord/voltha-protos/v5/go/voltha"
+
+ fu "github.com/opencord/voltha-lib-go/v7/pkg/flows"
+ "github.com/opencord/voltha-lib-go/v7/pkg/log"
rsrcMgr "github.com/opencord/voltha-openolt-adapter/internal/pkg/resourcemanager"
- ofp "github.com/opencord/voltha-protos/v4/go/openflow_13"
- openoltpb2 "github.com/opencord/voltha-protos/v4/go/openolt"
- tp_pb "github.com/opencord/voltha-protos/v4/go/tech_profile"
+ ofp "github.com/opencord/voltha-protos/v5/go/openflow_13"
+ openoltpb2 "github.com/opencord/voltha-protos/v5/go/openolt"
+ tp_pb "github.com/opencord/voltha-protos/v5/go/tech_profile"
)
var flowMgr []*OpenOltFlowMgr
diff --git a/internal/pkg/core/openolt_groupmgr.go b/internal/pkg/core/openolt_groupmgr.go
index 2bbbc3a..ff51041 100644
--- a/internal/pkg/core/openolt_groupmgr.go
+++ b/internal/pkg/core/openolt_groupmgr.go
@@ -16,15 +16,16 @@
import (
"context"
- "github.com/opencord/voltha-lib-go/v6/pkg/flows"
- "github.com/opencord/voltha-lib-go/v6/pkg/log"
+ "sync"
+
+ "github.com/opencord/voltha-lib-go/v7/pkg/flows"
+ "github.com/opencord/voltha-lib-go/v7/pkg/log"
"github.com/opencord/voltha-openolt-adapter/internal/pkg/olterrors"
rsrcMgr "github.com/opencord/voltha-openolt-adapter/internal/pkg/resourcemanager"
- ofp "github.com/opencord/voltha-protos/v4/go/openflow_13"
- openoltpb2 "github.com/opencord/voltha-protos/v4/go/openolt"
+ ofp "github.com/opencord/voltha-protos/v5/go/openflow_13"
+ openoltpb2 "github.com/opencord/voltha-protos/v5/go/openolt"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
- "sync"
)
//QueueInfoBrief has information about gemPortID and service priority associated with Mcast group
diff --git a/internal/pkg/core/openolt_test.go b/internal/pkg/core/openolt_test.go
index 7056063..c9486b6 100644
--- a/internal/pkg/core/openolt_test.go
+++ b/internal/pkg/core/openolt_test.go
@@ -28,28 +28,25 @@
"reflect"
"testing"
- conf "github.com/opencord/voltha-lib-go/v6/pkg/config"
+ conf "github.com/opencord/voltha-lib-go/v7/pkg/config"
+ vgrpc "github.com/opencord/voltha-lib-go/v7/pkg/grpc"
- com "github.com/opencord/voltha-lib-go/v6/pkg/adapters/common"
- "github.com/opencord/voltha-lib-go/v6/pkg/events"
- fu "github.com/opencord/voltha-lib-go/v6/pkg/flows"
- "github.com/opencord/voltha-lib-go/v6/pkg/kafka"
- "github.com/opencord/voltha-lib-go/v6/pkg/log"
+ "github.com/opencord/voltha-lib-go/v7/pkg/events"
+ fu "github.com/opencord/voltha-lib-go/v7/pkg/flows"
+ "github.com/opencord/voltha-lib-go/v7/pkg/log"
"github.com/opencord/voltha-openolt-adapter/internal/pkg/config"
"github.com/opencord/voltha-openolt-adapter/internal/pkg/olterrors"
- ic "github.com/opencord/voltha-protos/v4/go/inter_container"
- "github.com/opencord/voltha-protos/v4/go/openflow_13"
- ofp "github.com/opencord/voltha-protos/v4/go/openflow_13"
- "github.com/opencord/voltha-protos/v4/go/voltha"
+ ic "github.com/opencord/voltha-protos/v5/go/inter_container"
+ "github.com/opencord/voltha-protos/v5/go/openflow_13"
+ ofp "github.com/opencord/voltha-protos/v5/go/openflow_13"
+ "github.com/opencord/voltha-protos/v5/go/voltha"
)
// mocks the OpenOLT struct.
type fields struct {
deviceHandlers map[string]*DeviceHandler
- coreProxy *com.CoreProxy
- adapterProxy *com.AdapterProxy
+ coreClient *vgrpc.Client
eventProxy *events.EventProxy
- kafkaICProxy kafka.InterContainerProxy
numOnus int
KVStoreAddress string
KVStoreType string
@@ -70,10 +67,7 @@
func testOltObject(testOlt *fields) *OpenOLT {
return &OpenOLT{
deviceHandlers: testOlt.deviceHandlers,
- coreProxy: testOlt.coreProxy,
- adapterProxy: testOlt.adapterProxy,
eventProxy: testOlt.eventProxy,
- kafkaICProxy: testOlt.kafkaICProxy,
numOnus: testOlt.numOnus,
KVStoreAddress: testOlt.KVStoreAddress,
KVStoreType: testOlt.KVStoreType,
@@ -112,7 +106,7 @@
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
- if got := NewOpenOLT(tt.fields.ctx, tt.fields.kafkaICProxy, tt.fields.coreProxy, tt.fields.adapterProxy,
+ if got := NewOpenOLT(tt.fields.ctx, tt.fields.coreClient,
tt.fields.eventProxy, tt.configFlags, tt.cm); reflect.TypeOf(got) != reflect.TypeOf(tt.want) && got != nil {
t.Errorf("NewOpenOLT() error = %v, wantErr %v", got, tt.want)
}
@@ -120,34 +114,9 @@
}
}
-func TestOpenOLT_Abandon_device(t *testing.T) {
+func TestOpenOLT_ActivateImageUpdate(t *testing.T) {
type args struct {
- device *voltha.Device
- }
- tests := []struct {
- name string
- fields *fields
- args args
- wantErr error
- }{
- {"abandon_device-1", &fields{}, args{}, olterrors.ErrNotImplemented},
- {"abandon_device-2", &fields{}, args{}, olterrors.ErrNotImplemented},
- {"abandon_device-3", &fields{}, args{}, olterrors.ErrNotImplemented},
- }
- for _, tt := range tests {
- t.Run(tt.name, func(t *testing.T) {
- oo := testOltObject(tt.fields)
- if err := oo.Abandon_device(context.Background(), tt.args.device); err != tt.wantErr {
- t.Errorf("Abandon_device() error = %v, wantErr %v", err, tt.wantErr)
- }
- })
- }
-}
-
-func TestOpenOLT_Activate_image_update(t *testing.T) {
- type args struct {
- device *voltha.Device
- request *voltha.ImageDownload
+ request *ic.ImageDownloadMessage
}
tests := []struct {
name string
@@ -166,7 +135,7 @@
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
oo := testOltObject(tt.fields)
- got, err := oo.Activate_image_update(context.Background(), tt.args.device, tt.args.request)
+ got, err := oo.ActivateImageUpdate(context.Background(), tt.args.request)
if err != tt.wantErr && got == nil {
t.Errorf("Activate_image_update() error = %v, wantErr %v", err, tt.wantErr)
}
@@ -174,27 +143,7 @@
}
}
-func TestOpenOLT_Adapter_descriptor(t *testing.T) {
- tests := []struct {
- name string
- fields *fields
- wantErr error
- }{
- {"adapter_descriptor-1", &fields{}, olterrors.ErrNotImplemented},
- {"adapter_descriptor-2", &fields{}, olterrors.ErrNotImplemented},
- {"adapter_descriptor-3", &fields{}, olterrors.ErrNotImplemented},
- }
- for _, tt := range tests {
- t.Run(tt.name, func(t *testing.T) {
- oo := testOltObject(tt.fields)
- if err := oo.Adapter_descriptor(context.Background()); err != tt.wantErr {
- t.Errorf("Adapter_descriptor() error = %v, wantErr %v", err, tt.wantErr)
- }
- })
- }
-}
-
-func TestOpenOLT_Adopt_device(t *testing.T) {
+func TestOpenOLT_AdoptDevice(t *testing.T) {
type args struct {
device *voltha.Device
}
@@ -214,7 +163,7 @@
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
oo := testOltObject(tt.fields)
- err := oo.Adopt_device(context.Background(), tt.args.device)
+ _, err := oo.AdoptDevice(context.Background(), tt.args.device)
if (err != nil) && (reflect.TypeOf(err) !=
reflect.TypeOf(tt.wantErr)) && (tt.args.device == nil) {
t.Errorf("Adopt_device() error = %v, wantErr %v", err, tt.wantErr)
@@ -226,10 +175,9 @@
}
}
-func TestOpenOLT_Cancel_image_download(t *testing.T) {
+func TestOpenOLT_CancelImageDownload(t *testing.T) {
type args struct {
- device *voltha.Device
- request *voltha.ImageDownload
+ request *ic.ImageDownloadMessage
}
tests := []struct {
name string
@@ -248,7 +196,7 @@
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
oo := testOltObject(tt.fields)
- got, err := oo.Cancel_image_download(context.Background(), tt.args.device, tt.args.request)
+ got, err := oo.CancelImageDownload(context.Background(), tt.args.request)
if err != tt.wantErr && got == nil {
t.Errorf("Cancel_image_download() error = %v, wantErr %v", err, tt.wantErr)
}
@@ -256,7 +204,7 @@
}
}
-func TestOpenOLT_Delete_device(t *testing.T) {
+func TestOpenOLT_DeleteDevice(t *testing.T) {
type args struct {
device *voltha.Device
}
@@ -272,39 +220,14 @@
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
oo := testOltObject(tt.fields)
- if err := oo.Delete_device(context.Background(), tt.args.device); !reflect.DeepEqual(err, tt.wantErr) {
+ if _, err := oo.DeleteDevice(context.Background(), tt.args.device); !reflect.DeepEqual(err, tt.wantErr) {
t.Errorf("Delete_device() error = %v, wantErr %v", err, tt.wantErr)
}
})
}
}
-func TestOpenOLT_Device_types(t *testing.T) {
- tests := []struct {
- name string
- fields *fields
- want *voltha.DeviceTypes
- wantErr error
- }{
- {"device_types-1", &fields{}, &voltha.DeviceTypes{},
- olterrors.ErrNotImplemented},
- {"device_types-2", &fields{}, &voltha.DeviceTypes{},
- olterrors.ErrNotImplemented},
- {"device_types-3", &fields{}, &voltha.DeviceTypes{},
- olterrors.ErrNotImplemented},
- }
- for _, tt := range tests {
- t.Run(tt.name, func(t *testing.T) {
- oo := testOltObject(tt.fields)
- got, err := oo.Device_types(context.Background())
- if err != tt.wantErr && got == nil {
- t.Errorf("Device_types() error = %v, wantErr %v", err, tt.wantErr)
- }
- })
- }
-}
-
-func TestOpenOLT_Disable_device(t *testing.T) {
+func TestOpenOLT_DisableDevice(t *testing.T) {
type args struct {
device *voltha.Device
}
@@ -321,17 +244,16 @@
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
oo := testOltObject(tt.fields)
- if err := oo.Disable_device(context.Background(), tt.args.device); !reflect.DeepEqual(err, tt.wantErr) {
+ if _, err := oo.DisableDevice(context.Background(), tt.args.device); !reflect.DeepEqual(err, tt.wantErr) {
t.Errorf("Disable_device() error = %v, wantErr %v", err, tt.wantErr)
}
})
}
}
-func TestOpenOLT_Download_image(t *testing.T) {
+func TestOpenOLT_DownloadImage(t *testing.T) {
type args struct {
- device *voltha.Device
- request *voltha.ImageDownload
+ request *ic.ImageDownloadMessage
}
tests := []struct {
name string
@@ -350,7 +272,7 @@
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
oo := testOltObject(tt.fields)
- got, err := oo.Download_image(context.Background(), tt.args.device, tt.args.request)
+ got, err := oo.DownloadImage(context.Background(), tt.args.request)
if err != tt.wantErr && got == nil {
t.Errorf("Download_image() error = %v, wantErr %v", err, tt.wantErr)
}
@@ -358,34 +280,9 @@
}
}
-func TestOpenOLT_Get_device_details(t *testing.T) {
+func TestOpenOLT_GetImageDownloadStatus(t *testing.T) {
type args struct {
- device *voltha.Device
- }
- tests := []struct {
- name string
- fields *fields
- args args
- wantErr error
- }{
- {"get_device_details-1", &fields{}, args{}, olterrors.ErrNotImplemented},
- {"get_device_details-2", &fields{}, args{}, olterrors.ErrNotImplemented},
- {"get_device_details-3", &fields{}, args{}, olterrors.ErrNotImplemented},
- }
- for _, tt := range tests {
- t.Run(tt.name, func(t *testing.T) {
- oo := testOltObject(tt.fields)
- if err := oo.Get_device_details(context.Background(), tt.args.device); err != tt.wantErr {
- t.Errorf("Get_device_details() error = %v, wantErr %v", err, tt.wantErr)
- }
- })
- }
-}
-
-func TestOpenOLT_Get_image_download_status(t *testing.T) {
- type args struct {
- device *voltha.Device
- request *voltha.ImageDownload
+ request *ic.ImageDownloadMessage
}
tests := []struct {
name string
@@ -404,7 +301,7 @@
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
oo := testOltObject(tt.fields)
- got, err := oo.Get_image_download_status(context.Background(), tt.args.device, tt.args.request)
+ got, err := oo.GetImageDownloadStatus(context.Background(), tt.args.request)
if err != tt.wantErr && got == nil {
t.Errorf("Get_image_download_status() got = %v want = %v error = %v, wantErr %v",
got, tt.want, err, tt.wantErr)
@@ -413,7 +310,7 @@
}
}
-func TestOpenOLT_Get_ofp_device_info(t *testing.T) {
+func TestOpenOLT_GetOfpDeviceInfo(t *testing.T) {
type args struct {
device *voltha.Device
}
@@ -442,7 +339,7 @@
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
oo := testOltObject(tt.fields)
- got, err := oo.Get_ofp_device_info(context.Background(), tt.args.device)
+ got, err := oo.GetOfpDeviceInfo(context.Background(), tt.args.device)
if !reflect.DeepEqual(err, tt.wantErr) || !reflect.DeepEqual(got, tt.want) {
t.Errorf("Get_ofp_device_info() got = %v want = %v error = %v, wantErr = %v",
got, tt.want, err, tt.wantErr)
@@ -451,86 +348,64 @@
}
}
-func TestOpenOLT_Health(t *testing.T) {
- tests := []struct {
- name string
- fields *fields
- want *voltha.HealthStatus
- wantErr error
- }{
- {"health-1", &fields{}, &voltha.HealthStatus{}, olterrors.ErrNotImplemented},
- {"health-2", &fields{}, &voltha.HealthStatus{}, olterrors.ErrNotImplemented},
- {"health-3", &fields{}, &voltha.HealthStatus{}, olterrors.ErrNotImplemented},
- }
- for _, tt := range tests {
- t.Run(tt.name, func(t *testing.T) {
- oo := testOltObject(tt.fields)
- got, err := oo.Health(context.Background())
- if err != tt.wantErr && got == nil {
- t.Errorf("Get_ofp_port_info() error = %v, wantErr %v", err, tt.wantErr)
- }
- })
- }
-}
+// func TestOpenOLT_Process_inter_adapter_message(t *testing.T) {
+// type args struct {
+// msg *ic.InterAdapterMessage
+// }
+// var message1 = args{
+// msg: &ic.InterAdapterMessage{
+// Header: &ic.InterAdapterHeader{
+// Id: "olt",
+// ProxyDeviceId: "",
+// ToDeviceId: "onu1",
+// },
+// },
+// }
+// var message2 = args{
+// msg: &ic.InterAdapterMessage{
+// Header: &ic.InterAdapterHeader{
+// Id: "olt",
+// ProxyDeviceId: "olt",
+// ToDeviceId: "olt",
+// Type: ic.InterAdapterMessageType_OMCI_REQUEST,
+// },
+// },
+// }
+// var message3 = args{
+// msg: &ic.InterAdapterMessage{
+// Header: &ic.InterAdapterHeader{
+// Id: "olt",
+// ProxyDeviceId: "olt",
+// ToDeviceId: "olt",
+// Type: ic.InterAdapterMessageType_FLOW_REQUEST,
+// },
+// },
+// }
+// tests := []struct {
+// name string
+// fields *fields
+// args args
+// wantErrType reflect.Type
+// }{
+// {"process_inter_adaptor_messgae-1", mockOlt(), message1,
+// reflect.TypeOf(&olterrors.ErrNotFound{})},
+// {"process_inter_adaptor_messgae-2", mockOlt(), message2,
+// reflect.TypeOf(&olterrors.ErrAdapter{})},
+// {"process_inter_adaptor_messgae-3", mockOlt(), message3,
+// reflect.TypeOf(&olterrors.ErrInvalidValue{})},
+// }
+// for _, tt := range tests {
+// t.Run(tt.name, func(t *testing.T) {
+// oo := testOltObject(tt.fields)
+// if err := oo.Process_inter_adapter_message(context.Background(), tt.args.msg); reflect.TypeOf(err) != tt.wantErrType {
+// t.Errorf("Process_inter_adapter_message() error = %v, wantErr %v",
+// reflect.TypeOf(err), tt.wantErrType)
+// }
+// })
+// }
+// }
-func TestOpenOLT_Process_inter_adapter_message(t *testing.T) {
- type args struct {
- msg *ic.InterAdapterMessage
- }
- var message1 = args{
- msg: &ic.InterAdapterMessage{
- Header: &ic.InterAdapterHeader{
- Id: "olt",
- ProxyDeviceId: "",
- ToDeviceId: "onu1",
- },
- },
- }
- var message2 = args{
- msg: &ic.InterAdapterMessage{
- Header: &ic.InterAdapterHeader{
- Id: "olt",
- ProxyDeviceId: "olt",
- ToDeviceId: "olt",
- Type: ic.InterAdapterMessageType_OMCI_REQUEST,
- },
- },
- }
- var message3 = args{
- msg: &ic.InterAdapterMessage{
- Header: &ic.InterAdapterHeader{
- Id: "olt",
- ProxyDeviceId: "olt",
- ToDeviceId: "olt",
- Type: ic.InterAdapterMessageType_FLOW_REQUEST,
- },
- },
- }
- tests := []struct {
- name string
- fields *fields
- args args
- wantErrType reflect.Type
- }{
- {"process_inter_adaptor_messgae-1", mockOlt(), message1,
- reflect.TypeOf(&olterrors.ErrNotFound{})},
- {"process_inter_adaptor_messgae-2", mockOlt(), message2,
- reflect.TypeOf(&olterrors.ErrAdapter{})},
- {"process_inter_adaptor_messgae-3", mockOlt(), message3,
- reflect.TypeOf(&olterrors.ErrInvalidValue{})},
- }
- for _, tt := range tests {
- t.Run(tt.name, func(t *testing.T) {
- oo := testOltObject(tt.fields)
- if err := oo.Process_inter_adapter_message(context.Background(), tt.args.msg); reflect.TypeOf(err) != tt.wantErrType {
- t.Errorf("Process_inter_adapter_message() error = %v, wantErr %v",
- reflect.TypeOf(err), tt.wantErrType)
- }
- })
- }
-}
-
-func TestOpenOLT_Reboot_device(t *testing.T) {
+func TestOpenOLT_RebootDevice(t *testing.T) {
type args struct {
device *voltha.Device
}
@@ -547,14 +422,14 @@
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
oo := testOltObject(tt.fields)
- if err := oo.Reboot_device(context.Background(), tt.args.device); !reflect.DeepEqual(err, tt.wantErr) {
+ if _, err := oo.RebootDevice(context.Background(), tt.args.device); !reflect.DeepEqual(err, tt.wantErr) {
t.Errorf("Reboot_device() error = %v, wantErr %v", err, tt.wantErr)
}
})
}
}
-func TestOpenOLT_Receive_packet_out(t *testing.T) {
+func TestOpenOLT_SendPacketOut(t *testing.T) {
acts := []*ofp.OfpAction{
fu.SetField(fu.Metadata_ofp(uint64(ofp.OfpInstructionType_OFPIT_WRITE_METADATA))),
fu.SetField(fu.VlanVid(uint32(ofp.OfpVlanId_OFPVID_PRESENT) | 101)),
@@ -581,14 +456,17 @@
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
oo := testOltObject(tt.fields)
- if err := oo.Receive_packet_out(context.Background(), tt.args.deviceID, tt.args.egressPortNo, tt.args.packet); !reflect.DeepEqual(err, tt.wantErr) {
+ if _, err := oo.SendPacketOut(context.Background(), &ic.PacketOut{
+ DeviceId: tt.args.deviceID,
+ EgressPortNo: uint32(tt.args.egressPortNo),
+ Packet: tt.args.packet}); !reflect.DeepEqual(err, tt.wantErr) {
t.Errorf("Receive_packet_out() error = %v, wantErr %v", err, tt.wantErr)
}
})
}
}
-func TestOpenOLT_Reconcile_device(t *testing.T) {
+func TestOpenOLT_ReconcileDevice(t *testing.T) {
type args struct {
device *voltha.Device
}
@@ -606,14 +484,14 @@
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
oo := testOltObject(tt.fields)
- if err := oo.Reconcile_device(context.Background(), tt.args.device); !reflect.DeepEqual(err, tt.wantErr) {
+ if _, err := oo.ReconcileDevice(context.Background(), tt.args.device); !reflect.DeepEqual(err, tt.wantErr) {
t.Errorf("Reconcile_device() error = %v, wantErr %v", err, tt.wantErr)
}
})
}
}
-func TestOpenOLT_Reenable_device(t *testing.T) {
+func TestOpenOLT_ReEnableDevice(t *testing.T) {
type args struct {
device *voltha.Device
}
@@ -630,17 +508,16 @@
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
oo := testOltObject(tt.fields)
- if err := oo.Reenable_device(context.Background(), tt.args.device); !reflect.DeepEqual(err, tt.wantErr) {
+ if _, err := oo.ReEnableDevice(context.Background(), tt.args.device); !reflect.DeepEqual(err, tt.wantErr) {
t.Errorf("Reenable_device() error = %v, wantErr %v", err, tt.wantErr)
}
})
}
}
-func TestOpenOLT_Revert_image_update(t *testing.T) {
+func TestOpenOLT_RevertImageUpdate(t *testing.T) {
type args struct {
- device *voltha.Device
- request *voltha.ImageDownload
+ request *ic.ImageDownloadMessage
}
tests := []struct {
name string
@@ -659,7 +536,7 @@
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
oo := testOltObject(tt.fields)
- got, err := oo.Revert_image_update(context.Background(), tt.args.device, tt.args.request)
+ got, err := oo.RevertImageUpdate(context.Background(), tt.args.request)
if err != tt.wantErr && got == nil {
t.Log("error :", err)
}
@@ -667,7 +544,7 @@
}
}
-func TestOpenOLT_Self_test_device(t *testing.T) {
+func TestOpenOLT_SelfTestDevice(t *testing.T) {
type args struct {
device *voltha.Device
}
@@ -684,7 +561,7 @@
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
oo := testOltObject(tt.fields)
- if err := oo.Self_test_device(context.Background(), tt.args.device); err != tt.wantErr {
+ if _, err := oo.SelfTestDevice(context.Background(), tt.args.device); err != tt.wantErr {
t.Errorf("Self_test_device() error = %v, wantErr %v", err, tt.wantErr)
}
})
@@ -738,7 +615,7 @@
}
}
-func TestOpenOLT_Suppress_event(t *testing.T) {
+func TestOpenOLT_SuppressEvent(t *testing.T) {
type args struct {
filter *voltha.EventFilter
}
@@ -755,14 +632,14 @@
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
oo := testOltObject(tt.fields)
- if err := oo.Suppress_event(context.Background(), tt.args.filter); err != tt.wantErr {
+ if _, err := oo.SuppressEvent(context.Background(), tt.args.filter); err != tt.wantErr {
t.Errorf("Suppress_event() error = %v, wantErr %v", err, tt.wantErr)
}
})
}
}
-func TestOpenOLT_Unsuppress_event(t *testing.T) {
+func TestOpenOLT_UnSuppressEvent(t *testing.T) {
type args struct {
filter *voltha.EventFilter
}
@@ -779,14 +656,14 @@
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
oo := testOltObject(tt.fields)
- if err := oo.Unsuppress_event(context.Background(), tt.args.filter); err != tt.wantErr {
+ if _, err := oo.UnSuppressEvent(context.Background(), tt.args.filter); err != tt.wantErr {
t.Errorf("Unsuppress_event() error = %v, wantErr %v", err, tt.wantErr)
}
})
}
}
-func TestOpenOLT_Update_flows_bulk(t *testing.T) {
+func TestOpenOLT_UpdateFlowsBulk(t *testing.T) {
type args struct {
device *voltha.Device
flows *voltha.Flows
@@ -806,14 +683,19 @@
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
oo := testOltObject(tt.fields)
- if err := oo.Update_flows_bulk(context.Background(), tt.args.device, tt.args.flows, tt.args.groups, tt.args.flowMetadata); err != tt.wantErr {
+ if _, err := oo.UpdateFlowsBulk(context.Background(), &ic.BulkFlows{
+ Device: tt.args.device,
+ Flows: tt.args.flows,
+ Groups: tt.args.groups,
+ FlowMetadata: tt.args.flowMetadata,
+ }); err != tt.wantErr {
t.Errorf("Update_flows_bulk() error = %v, wantErr %v", err, tt.wantErr)
}
})
}
}
-func TestOpenOLT_Update_flows_incrementally(t *testing.T) {
+func TestOpenOLT_UpdateFlowsIncrementally(t *testing.T) {
type args struct {
device *voltha.Device
flows *openflow_13.FlowChanges
@@ -834,14 +716,18 @@
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
oo := testOltObject(tt.fields)
- if err := oo.Update_flows_incrementally(context.Background(), tt.args.device, tt.args.flows, tt.args.groups, tt.args.flowMetadata); !reflect.DeepEqual(err, tt.wantErr) {
+ if _, err := oo.UpdateFlowsIncrementally(context.Background(), &ic.IncrementalFlows{
+ Device: tt.args.device,
+ Flows: tt.args.flows,
+ Groups: tt.args.groups,
+ FlowMetadata: tt.args.flowMetadata}); !reflect.DeepEqual(err, tt.wantErr) {
t.Errorf("Update_flows_incrementally() error = %v, wantErr %v", err, tt.wantErr)
}
})
}
}
-func TestOpenOLT_Update_pm_config(t *testing.T) {
+func TestOpenOLT_UpdatePmConfig(t *testing.T) {
type args struct {
device *voltha.Device
pmConfigs *voltha.PmConfigs
@@ -858,8 +744,7 @@
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
oo := testOltObject(tt.fields)
-
- if err := oo.Update_pm_config(context.Background(), tt.args.device, tt.args.pmConfigs); !reflect.DeepEqual(err, tt.wantErr) {
+ if _, err := oo.UpdatePmConfig(context.Background(), &ic.PmConfigsInfo{DeviceId: tt.args.device.Id, PmConfigs: tt.args.pmConfigs}); !reflect.DeepEqual(err, tt.wantErr) {
t.Errorf("Update_pm_config() error = %v, wantErr %v", err, tt.wantErr)
}
@@ -889,10 +774,9 @@
}
}
-func TestOpenOLT_Enable_port(t *testing.T) {
+func TestOpenOLT_EnablePort(t *testing.T) {
type args struct {
- deviceID string
- port *voltha.Port
+ port *voltha.Port
}
tests := []struct {
name string
@@ -901,23 +785,22 @@
wantErr bool
}{
// TODO: Add test cases.
- {"Enable_port-1", mockOlt(), args{deviceID: "olt", port: &voltha.Port{Type: voltha.Port_PON_OLT, PortNo: 1}}, false},
- {"Enable_port-2", mockOlt(), args{deviceID: "olt", port: &voltha.Port{Type: voltha.Port_ETHERNET_NNI, PortNo: 1}}, true},
+ {"Enable_port-1", mockOlt(), args{port: &voltha.Port{Type: voltha.Port_PON_OLT, PortNo: 1, DeviceId: "olt"}}, false},
+ {"Enable_port-2", mockOlt(), args{port: &voltha.Port{Type: voltha.Port_ETHERNET_NNI, PortNo: 1, DeviceId: "olt"}}, true},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
oo := testOltObject(tt.fields)
- if err := oo.Enable_port(context.Background(), tt.args.deviceID, tt.args.port); (err != nil) != tt.wantErr {
+ if _, err := oo.EnablePort(context.Background(), tt.args.port); (err != nil) != tt.wantErr {
t.Errorf("OpenOLT.Enable_port() error = %v, wantErr %v", err, tt.wantErr)
}
})
}
}
-func TestOpenOLT_Disable_port(t *testing.T) {
+func TestOpenOLT_DisablePort(t *testing.T) {
type args struct {
- deviceID string
- port *voltha.Port
+ port *voltha.Port
}
tests := []struct {
name string
@@ -926,13 +809,13 @@
wantErr bool
}{
// TODO: Add test cases.
- {"Disable_port-1", mockOlt(), args{deviceID: "olt", port: &voltha.Port{Type: voltha.Port_PON_OLT, PortNo: 1}}, false},
- {"Disable_port-2", mockOlt(), args{deviceID: "olt", port: &voltha.Port{Type: voltha.Port_ETHERNET_NNI, PortNo: 1}}, true},
+ {"Disable_port-1", mockOlt(), args{port: &voltha.Port{Type: voltha.Port_PON_OLT, PortNo: 1, DeviceId: "olt"}}, false},
+ {"Disable_port-2", mockOlt(), args{port: &voltha.Port{Type: voltha.Port_ETHERNET_NNI, PortNo: 1, DeviceId: "olt"}}, true},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
oo := testOltObject(tt.fields)
- if err := oo.Disable_port(context.Background(), tt.args.deviceID, tt.args.port); (err != nil) != tt.wantErr {
+ if _, err := oo.DisablePort(context.Background(), tt.args.port); (err != nil) != tt.wantErr {
t.Errorf("OpenOLT.Disable_port() error = %v, wantErr %v", err, tt.wantErr)
}
})
diff --git a/internal/pkg/core/statsmanager.go b/internal/pkg/core/statsmanager.go
index 8b07fa2..2257425 100755
--- a/internal/pkg/core/statsmanager.go
+++ b/internal/pkg/core/statsmanager.go
@@ -21,15 +21,16 @@
"container/list"
"context"
"fmt"
- rsrcMgr "github.com/opencord/voltha-openolt-adapter/internal/pkg/resourcemanager"
"sync"
"time"
- "github.com/opencord/voltha-lib-go/v6/pkg/log"
+ rsrcMgr "github.com/opencord/voltha-openolt-adapter/internal/pkg/resourcemanager"
+
+ "github.com/opencord/voltha-lib-go/v7/pkg/log"
"github.com/opencord/voltha-openolt-adapter/internal/pkg/olterrors"
- "github.com/opencord/voltha-protos/v4/go/extension"
- "github.com/opencord/voltha-protos/v4/go/openolt"
- "github.com/opencord/voltha-protos/v4/go/voltha"
+ "github.com/opencord/voltha-protos/v5/go/extension"
+ "github.com/opencord/voltha-protos/v5/go/openolt"
+ "github.com/opencord/voltha-protos/v5/go/voltha"
)
const (
diff --git a/internal/pkg/core/statsmanager_test.go b/internal/pkg/core/statsmanager_test.go
index 1ea2ddd..219e255 100644
--- a/internal/pkg/core/statsmanager_test.go
+++ b/internal/pkg/core/statsmanager_test.go
@@ -23,8 +23,8 @@
"reflect"
"testing"
- "github.com/opencord/voltha-protos/v4/go/openolt"
- "github.com/opencord/voltha-protos/v4/go/voltha"
+ "github.com/opencord/voltha-protos/v5/go/openolt"
+ "github.com/opencord/voltha-protos/v5/go/voltha"
)
func TestOpenOltStatisticsMgr_PortStatisticsIndication(t *testing.T) {
diff --git a/internal/pkg/olterrors/common.go b/internal/pkg/olterrors/common.go
index 9ac8676..2edb2a6 100644
--- a/internal/pkg/olterrors/common.go
+++ b/internal/pkg/olterrors/common.go
@@ -18,7 +18,7 @@
package olterrors
import (
- "github.com/opencord/voltha-lib-go/v6/pkg/log"
+ "github.com/opencord/voltha-lib-go/v7/pkg/log"
)
var logger log.CLogger
diff --git a/internal/pkg/olterrors/olterrors.go b/internal/pkg/olterrors/olterrors.go
index 4aa2373..bafc721 100644
--- a/internal/pkg/olterrors/olterrors.go
+++ b/internal/pkg/olterrors/olterrors.go
@@ -21,8 +21,9 @@
"context"
"encoding/json"
"fmt"
- "github.com/opencord/voltha-lib-go/v6/pkg/log"
"strings"
+
+ "github.com/opencord/voltha-lib-go/v7/pkg/log"
)
const (
diff --git a/internal/pkg/resourcemanager/common.go b/internal/pkg/resourcemanager/common.go
index 187242c..77e217e 100644
--- a/internal/pkg/resourcemanager/common.go
+++ b/internal/pkg/resourcemanager/common.go
@@ -18,7 +18,7 @@
package resourcemanager
import (
- "github.com/opencord/voltha-lib-go/v6/pkg/log"
+ "github.com/opencord/voltha-lib-go/v7/pkg/log"
)
var logger log.CLogger
diff --git a/internal/pkg/resourcemanager/resourcemanager.go b/internal/pkg/resourcemanager/resourcemanager.go
index 830732c..0f2a65e 100755
--- a/internal/pkg/resourcemanager/resourcemanager.go
+++ b/internal/pkg/resourcemanager/resourcemanager.go
@@ -22,17 +22,18 @@
"encoding/json"
"errors"
"fmt"
- tp "github.com/opencord/voltha-lib-go/v6/pkg/techprofile"
"strings"
"sync"
"time"
- "github.com/opencord/voltha-lib-go/v6/pkg/db"
- "github.com/opencord/voltha-lib-go/v6/pkg/db/kvstore"
- "github.com/opencord/voltha-lib-go/v6/pkg/log"
- ponrmgr "github.com/opencord/voltha-lib-go/v6/pkg/ponresourcemanager"
- ofp "github.com/opencord/voltha-protos/v4/go/openflow_13"
- "github.com/opencord/voltha-protos/v4/go/openolt"
+ tp "github.com/opencord/voltha-lib-go/v7/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"
+ ponrmgr "github.com/opencord/voltha-lib-go/v7/pkg/ponresourcemanager"
+ ofp "github.com/opencord/voltha-protos/v5/go/openflow_13"
+ "github.com/opencord/voltha-protos/v5/go/openolt"
)
const (
diff --git a/internal/pkg/resourcemanager/resourcemanager_test.go b/internal/pkg/resourcemanager/resourcemanager_test.go
index 9d8d068..4244977 100644
--- a/internal/pkg/resourcemanager/resourcemanager_test.go
+++ b/internal/pkg/resourcemanager/resourcemanager_test.go
@@ -27,21 +27,22 @@
"context"
"encoding/json"
"errors"
- "github.com/opencord/voltha-lib-go/v6/pkg/techprofile"
- "github.com/opencord/voltha-openolt-adapter/pkg/mocks"
"reflect"
"strconv"
"strings"
"testing"
"time"
- "github.com/opencord/voltha-lib-go/v6/pkg/db"
- "github.com/opencord/voltha-lib-go/v6/pkg/db/kvstore"
- fu "github.com/opencord/voltha-lib-go/v6/pkg/flows"
- "github.com/opencord/voltha-lib-go/v6/pkg/log"
- ponrmgr "github.com/opencord/voltha-lib-go/v6/pkg/ponresourcemanager"
- ofp "github.com/opencord/voltha-protos/v4/go/openflow_13"
- "github.com/opencord/voltha-protos/v4/go/openolt"
+ "github.com/opencord/voltha-lib-go/v7/pkg/techprofile"
+ "github.com/opencord/voltha-openolt-adapter/pkg/mocks"
+
+ "github.com/opencord/voltha-lib-go/v7/pkg/db"
+ "github.com/opencord/voltha-lib-go/v7/pkg/db/kvstore"
+ fu "github.com/opencord/voltha-lib-go/v7/pkg/flows"
+ "github.com/opencord/voltha-lib-go/v7/pkg/log"
+ ponrmgr "github.com/opencord/voltha-lib-go/v7/pkg/ponresourcemanager"
+ ofp "github.com/opencord/voltha-protos/v5/go/openflow_13"
+ "github.com/opencord/voltha-protos/v5/go/openolt"
)
func init() {