[VOL-1359] This commit consists of the creation of the simulated
OLT and ONU adapters (in Go language). This update also provides
the set of files to build and run these containers.
Change-Id: Id7b0c77fdf60cb02c39908d4374d3e93fab5de67
diff --git a/adapters/simulated_olt/adaptercore/device_handler.go b/adapters/simulated_olt/adaptercore/device_handler.go
new file mode 100644
index 0000000..0741761
--- /dev/null
+++ b/adapters/simulated_olt/adaptercore/device_handler.go
@@ -0,0 +1,196 @@
+/*
+ * Copyright 2018-present Open Networking Foundation
+
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+
+ * http://www.apache.org/licenses/LICENSE-2.0
+
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package adaptercore
+
+import (
+ "context"
+ "github.com/gogo/protobuf/proto"
+ com "github.com/opencord/voltha-go/adapters/common"
+ "github.com/opencord/voltha-go/common/log"
+ ic "github.com/opencord/voltha-go/protos/inter_container"
+ of "github.com/opencord/voltha-go/protos/openflow_13"
+ "github.com/opencord/voltha-go/protos/voltha"
+ "strconv"
+ "strings"
+ "sync"
+)
+
+//DeviceHandler follows the same patterns as ponsim_olt. The only difference is that it does not
+// interact with an OLT device.
+type DeviceHandler struct {
+ deviceId string
+ deviceType string
+ device *voltha.Device
+ coreProxy *com.CoreProxy
+ simulatedOLT *SimulatedOLT
+ nniPort *voltha.Port
+ ponPort *voltha.Port
+ exitChannel chan int
+ lockDevice sync.RWMutex
+}
+
+//NewDeviceHandler creates a new device handler
+func NewDeviceHandler(cp *com.CoreProxy, device *voltha.Device, adapter *SimulatedOLT) *DeviceHandler {
+ var dh DeviceHandler
+ dh.coreProxy = cp
+ cloned := (proto.Clone(device)).(*voltha.Device)
+ dh.deviceId = cloned.Id
+ dh.deviceType = cloned.Type
+ dh.device = cloned
+ dh.simulatedOLT = adapter
+ dh.exitChannel = make(chan int, 1)
+ dh.lockDevice = sync.RWMutex{}
+ return &dh
+}
+
+// start save the device to the data model
+func (dh *DeviceHandler) start(ctx context.Context) {
+ dh.lockDevice.Lock()
+ defer dh.lockDevice.Unlock()
+ log.Debugw("starting-device-agent", log.Fields{"device": dh.device})
+ // Add the initial device to the local model
+ log.Debug("device-agent-started")
+}
+
+// stop stops the device dh. Not much to do for now
+func (dh *DeviceHandler) stop(ctx context.Context) {
+ dh.lockDevice.Lock()
+ defer dh.lockDevice.Unlock()
+ log.Debug("stopping-device-agent")
+ dh.exitChannel <- 1
+ log.Debug("device-agent-stopped")
+}
+
+func macAddressToUint32Array(mac string) []uint32 {
+ slist := strings.Split(mac, ":")
+ result := make([]uint32, len(slist))
+ var err error
+ var tmp int64
+ for index, val := range slist {
+ if tmp, err = strconv.ParseInt(val, 16, 32); err != nil {
+ return []uint32{1, 2, 3, 4, 5, 6}
+ }
+ result[index] = uint32(tmp)
+ }
+ return result
+}
+
+func (dh *DeviceHandler) AdoptDevice(device *voltha.Device) {
+ log.Debugw("AdoptDevice", log.Fields{"deviceId": device.Id})
+
+ // Update the device info
+ cloned := proto.Clone(device).(*voltha.Device)
+ cloned.Root = true
+ cloned.Vendor = "simulators"
+ cloned.Model = "go-simulators"
+ cloned.SerialNumber = com.GetRandomSerialNumber()
+ cloned.MacAddress = strings.ToUpper(com.GetRandomMacAddress())
+
+ // Synchronous call to update device - this method is run in its own go routine
+ if err := dh.coreProxy.DeviceUpdate(nil, cloned); err != nil {
+ log.Errorw("error-updating-device", log.Fields{"deviceId": device.Id, "error": err})
+ }
+
+ // Now create the NNI Port
+ dh.nniPort = &voltha.Port{
+ PortNo: 2,
+ Label: "NNI facing Ethernet port",
+ Type: voltha.Port_ETHERNET_NNI,
+ OperStatus: voltha.OperStatus_ACTIVE,
+ }
+
+ // Synchronous call to update device - this method is run in its own go routine
+ if err := dh.coreProxy.PortCreated(nil, cloned.Id, dh.nniPort); err != nil {
+ log.Errorw("error-creating-nni-port", log.Fields{"deviceId": device.Id, "error": err})
+ }
+
+ // Now create the PON Port
+ dh.ponPort = &voltha.Port{
+ PortNo: 1,
+ Label: "PON port",
+ Type: voltha.Port_PON_OLT,
+ OperStatus: voltha.OperStatus_ACTIVE,
+ }
+
+ // Synchronous call to update device - this method is run in its own go routine
+ if err := dh.coreProxy.PortCreated(nil, cloned.Id, dh.ponPort); err != nil {
+ log.Errorw("error-creating-nni-port", log.Fields{"deviceId": device.Id, "error": err})
+ }
+
+ cloned.ConnectStatus = voltha.ConnectStatus_REACHABLE
+ cloned.OperStatus = voltha.OperStatus_ACTIVE
+
+ // Update the device state
+ if err := dh.coreProxy.DeviceStateUpdate(nil, cloned.Id, cloned.ConnectStatus, cloned.OperStatus); err != nil {
+ log.Errorw("error-creating-nni-port", log.Fields{"deviceId": device.Id, "error": err})
+ }
+
+ // Register Child device
+ initialUniPortNo := 100
+ log.Debugw("registering-onus", log.Fields{"total": dh.simulatedOLT.numOnus})
+ for i := 0; i < dh.simulatedOLT.numOnus; i++ {
+ go dh.coreProxy.ChildDeviceDetected(
+ nil,
+ cloned.Id,
+ 1,
+ "simulated_onu",
+ initialUniPortNo+i)
+ }
+ dh.device = cloned
+}
+
+func (dh *DeviceHandler) GetOfpDeviceInfo(device *voltha.Device) (*ic.SwitchCapability, error) {
+ return &ic.SwitchCapability{
+ Desc: &of.OfpDesc{
+ HwDesc: "simulated_pon",
+ SwDesc: "simulated_pon",
+ SerialNum: dh.device.SerialNumber,
+ },
+ SwitchFeatures: &of.OfpSwitchFeatures{
+ NBuffers: 256,
+ NTables: 2,
+ Capabilities: uint32(of.OfpCapabilities_OFPC_FLOW_STATS |
+ of.OfpCapabilities_OFPC_TABLE_STATS |
+ of.OfpCapabilities_OFPC_PORT_STATS |
+ of.OfpCapabilities_OFPC_GROUP_STATS),
+ },
+ }, nil
+}
+
+func (dh *DeviceHandler) GetOfpPortInfo(device *voltha.Device, portNo int64) (*ic.PortCapability, error) {
+ cap := uint32(of.OfpPortFeatures_OFPPF_1GB_FD | of.OfpPortFeatures_OFPPF_FIBER)
+ return &ic.PortCapability{
+ Port: &voltha.LogicalPort{
+ OfpPort: &of.OfpPort{
+ HwAddr: macAddressToUint32Array(dh.device.MacAddress),
+ Config: 0,
+ State: uint32(of.OfpPortState_OFPPS_LIVE),
+ Curr: cap,
+ Advertised: cap,
+ Peer: cap,
+ CurrSpeed: uint32(of.OfpPortFeatures_OFPPF_1GB_FD),
+ MaxSpeed: uint32(of.OfpPortFeatures_OFPPF_1GB_FD),
+ },
+ DeviceId: dh.device.Id,
+ DevicePortNo: uint32(portNo),
+ },
+ }, nil
+}
+
+func (dh *DeviceHandler) Process_inter_adapter_message(msg *ic.InterAdapterMessage) error {
+ log.Debugw("Process_inter_adapter_message", log.Fields{"msgId": msg.Header.Id})
+ return nil
+}
diff --git a/adapters/simulated_olt/adaptercore/simulated_olt.go b/adapters/simulated_olt/adaptercore/simulated_olt.go
new file mode 100644
index 0000000..bb6883a
--- /dev/null
+++ b/adapters/simulated_olt/adaptercore/simulated_olt.go
@@ -0,0 +1,243 @@
+/*
+ * Copyright 2018-present Open Networking Foundation
+
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+
+ * http://www.apache.org/licenses/LICENSE-2.0
+
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package adaptercore
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ com "github.com/opencord/voltha-go/adapters/common"
+ "github.com/opencord/voltha-go/common/log"
+ "github.com/opencord/voltha-go/kafka"
+ ic "github.com/opencord/voltha-go/protos/inter_container"
+ "github.com/opencord/voltha-go/protos/openflow_13"
+ "github.com/opencord/voltha-go/protos/voltha"
+ "sync"
+)
+
+type SimulatedOLT struct {
+ deviceHandlers map[string]*DeviceHandler
+ coreProxy *com.CoreProxy
+ kafkaICProxy *kafka.InterContainerProxy
+ numOnus int
+ exitChannel chan int
+ lockDeviceHandlersMap sync.RWMutex
+}
+
+func NewSimulatedOLT(ctx context.Context, kafkaICProxy *kafka.InterContainerProxy, coreProxy *com.CoreProxy, onuNumber int) *SimulatedOLT {
+ var simulatedOLT SimulatedOLT
+ simulatedOLT.exitChannel = make(chan int, 1)
+ simulatedOLT.deviceHandlers = make(map[string]*DeviceHandler)
+ simulatedOLT.kafkaICProxy = kafkaICProxy
+ simulatedOLT.numOnus = onuNumber
+ simulatedOLT.coreProxy = coreProxy
+ simulatedOLT.lockDeviceHandlersMap = sync.RWMutex{}
+ return &simulatedOLT
+}
+
+func (so *SimulatedOLT) Start(ctx context.Context) error {
+ log.Info("starting-device-manager")
+ log.Info("device-manager-started")
+ return nil
+}
+
+func (so *SimulatedOLT) Stop(ctx context.Context) error {
+ log.Info("stopping-device-manager")
+ so.exitChannel <- 1
+ log.Info("device-manager-stopped")
+ return nil
+}
+
+func sendResponse(ctx context.Context, ch chan interface{}, result interface{}) {
+ if ctx.Err() == nil {
+ // Returned response only of the ctx has not been cancelled/timeout/etc
+ // Channel is automatically closed when a context is Done
+ ch <- result
+ log.Debugw("sendResponse", log.Fields{"result": result})
+ } else {
+ // Should the transaction be reverted back?
+ log.Debugw("sendResponse-context-error", log.Fields{"context-error": ctx.Err()})
+ }
+}
+
+func (so *SimulatedOLT) addDeviceHandlerToMap(agent *DeviceHandler) {
+ so.lockDeviceHandlersMap.Lock()
+ defer so.lockDeviceHandlersMap.Unlock()
+ if _, exist := so.deviceHandlers[agent.deviceId]; !exist {
+ so.deviceHandlers[agent.deviceId] = agent
+ }
+}
+
+func (so *SimulatedOLT) deleteDeviceHandlerToMap(agent *DeviceHandler) {
+ so.lockDeviceHandlersMap.Lock()
+ defer so.lockDeviceHandlersMap.Unlock()
+ delete(so.deviceHandlers, agent.deviceId)
+}
+
+func (so *SimulatedOLT) getDeviceHandler(deviceId string) *DeviceHandler {
+ so.lockDeviceHandlersMap.Lock()
+ defer so.lockDeviceHandlersMap.Unlock()
+ if agent, ok := so.deviceHandlers[deviceId]; ok {
+ return agent
+ }
+ return nil
+}
+
+func (so *SimulatedOLT) createDeviceTopic(device *voltha.Device) error {
+ log.Infow("create-device-topic", log.Fields{"deviceId": device.Id})
+ deviceTopic := kafka.Topic{Name: so.kafkaICProxy.DefaultTopic.Name + "_" + device.Id}
+ if err := so.kafkaICProxy.SubscribeWithDefaultRequestHandler(deviceTopic); err != nil {
+ log.Infow("create-device-topic-failed", log.Fields{"deviceId": device.Id, "error": err})
+ return err
+ }
+ return nil
+}
+
+func (so *SimulatedOLT) Adopt_device(device *voltha.Device) error {
+ if device == nil {
+ log.Warn("device-is-nil")
+ return errors.New("nil-device")
+ }
+ log.Infow("adopt-device", log.Fields{"deviceId": device.Id})
+ var handler *DeviceHandler
+ if handler = so.getDeviceHandler(device.Id); handler == nil {
+ handler := NewDeviceHandler(so.coreProxy, device, so)
+ so.addDeviceHandlerToMap(handler)
+ go handler.AdoptDevice(device)
+ // Launch the creation of the device topic
+ go so.createDeviceTopic(device)
+ }
+ return nil
+}
+
+func (so *SimulatedOLT) Get_ofp_device_info(device *voltha.Device) (*ic.SwitchCapability, error) {
+ log.Infow("Get_ofp_device_info", log.Fields{"deviceId": device.Id})
+ if handler := so.getDeviceHandler(device.Id); handler != nil {
+ return handler.GetOfpDeviceInfo(device)
+ }
+ log.Errorw("device-handler-not-set", log.Fields{"deviceId": device.Id})
+ return nil, errors.New("device-handler-not-set")
+}
+
+func (so *SimulatedOLT) Get_ofp_port_info(device *voltha.Device, port_no int64) (*ic.PortCapability, error) {
+ log.Infow("Get_ofp_port_info", log.Fields{"deviceId": device.Id})
+ if handler := so.getDeviceHandler(device.Id); handler != nil {
+ return handler.GetOfpPortInfo(device, port_no)
+ }
+ log.Errorw("device-handler-not-set", log.Fields{"deviceId": device.Id})
+ return nil, errors.New("device-handler-not-set")
+}
+
+func (so *SimulatedOLT) Process_inter_adapter_message(msg *ic.InterAdapterMessage) error {
+ log.Infow("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 := so.getDeviceHandler(targetDevice); handler != nil {
+ return handler.Process_inter_adapter_message(msg)
+ }
+ return errors.New(fmt.Sprintf("handler-not-found-%s", targetDevice))
+}
+
+func (so *SimulatedOLT) Adapter_descriptor() error {
+ return errors.New("UnImplemented")
+}
+
+func (so *SimulatedOLT) Device_types() (*voltha.DeviceTypes, error) {
+ return nil, errors.New("UnImplemented")
+}
+
+func (so *SimulatedOLT) Health() (*voltha.HealthStatus, error) {
+ return nil, errors.New("UnImplemented")
+}
+
+func (so *SimulatedOLT) Reconcile_device(device *voltha.Device) error {
+ return errors.New("UnImplemented")
+}
+
+func (so *SimulatedOLT) Abandon_device(device *voltha.Device) error {
+ return errors.New("UnImplemented")
+}
+
+func (so *SimulatedOLT) Disable_device(device *voltha.Device) error {
+ return errors.New("UnImplemented")
+}
+
+func (so *SimulatedOLT) Reenable_device(device *voltha.Device) error {
+ return errors.New("UnImplemented")
+}
+
+func (so *SimulatedOLT) Reboot_device(device *voltha.Device) error {
+ return errors.New("UnImplemented")
+}
+
+func (so *SimulatedOLT) Self_test_device(device *voltha.Device) error {
+ return errors.New("UnImplemented")
+}
+
+func (so *SimulatedOLT) Gelete_device(device *voltha.Device) error {
+ return errors.New("UnImplemented")
+}
+
+func (so *SimulatedOLT) Get_device_details(device *voltha.Device) error {
+ return errors.New("UnImplemented")
+}
+
+func (so *SimulatedOLT) Update_flows_bulk(device *voltha.Device, flows *voltha.Flows, groups *voltha.FlowGroups) error {
+ return errors.New("UnImplemented")
+}
+
+func (so *SimulatedOLT) Update_flows_incrementally(device *voltha.Device, flows *openflow_13.FlowChanges, groups *openflow_13.FlowGroupChanges) error {
+ return errors.New("UnImplemented")
+}
+
+func (so *SimulatedOLT) Update_pm_config(device *voltha.Device, pm_configs *voltha.PmConfigs) error {
+ return errors.New("UnImplemented")
+}
+
+func (so *SimulatedOLT) Receive_packet_out(device *voltha.Device, egress_port_no int, msg openflow_13.PacketOut) error {
+ return errors.New("UnImplemented")
+}
+
+func (so *SimulatedOLT) Suppress_alarm(filter *voltha.AlarmFilter) error {
+ return errors.New("UnImplemented")
+}
+
+func (so *SimulatedOLT) Unsuppress_alarm(filter *voltha.AlarmFilter) error {
+ return errors.New("UnImplemented")
+}
+
+func (so *SimulatedOLT) Download_image(device *voltha.Device, request *voltha.ImageDownload) error {
+ return errors.New("UnImplemented")
+}
+
+func (so *SimulatedOLT) Get_image_download_status(device *voltha.Device, request *voltha.ImageDownload) error {
+ return errors.New("UnImplemented")
+}
+
+func (so *SimulatedOLT) Cancel_image_download(device *voltha.Device, request *voltha.ImageDownload) error {
+ return errors.New("UnImplemented")
+}
+
+func (so *SimulatedOLT) Activate_image_update(device *voltha.Device, request *voltha.ImageDownload) error {
+ return errors.New("UnImplemented")
+}
+
+func (so *SimulatedOLT) Revert_image_update(device *voltha.Device, request *voltha.ImageDownload) error {
+ return errors.New("UnImplemented")
+}
diff --git a/adapters/simulated_olt/config/config.go b/adapters/simulated_olt/config/config.go
new file mode 100644
index 0000000..5cf0dc0
--- /dev/null
+++ b/adapters/simulated_olt/config/config.go
@@ -0,0 +1,143 @@
+/*
+ * Copyright 2018-present Open Networking Foundation
+
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+
+ * http://www.apache.org/licenses/LICENSE-2.0
+
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package config
+
+import (
+ "flag"
+ "fmt"
+ "github.com/opencord/voltha-go/common/log"
+ "os"
+)
+
+// Simulated OLT default constants
+const (
+ EtcdStoreName = "etcd"
+ default_InstanceID = "simulatedOlt001"
+ default_KafkaAdapterHost = "127.0.0.1"
+ default_KafkaAdapterPort = 9092
+ default_KafkaClusterHost = "127.0.0.1"
+ default_KafkaClusterPort = 9094
+ default_KVStoreType = EtcdStoreName
+ default_KVStoreTimeout = 5 //in seconds
+ default_KVStoreHost = "127.0.0.1"
+ default_KVStorePort = 2379 // Consul = 8500; Etcd = 2379
+ default_LogLevel = 0
+ default_Banner = false
+ default_Topic = "simulated_olt"
+ default_CoreTopic = "rwcore"
+ default_OnuNumber = 1
+)
+
+// AdapterFlags represents the set of configurations used by the read-write adaptercore service
+type AdapterFlags struct {
+ // Command line parameters
+ InstanceID string
+ KafkaAdapterHost string
+ KafkaAdapterPort int
+ KafkaClusterHost string
+ KafkaClusterPort int
+ KVStoreType string
+ KVStoreTimeout int // in seconds
+ KVStoreHost string
+ KVStorePort int
+ Topic string
+ CoreTopic string
+ LogLevel int
+ OnuNumber int
+ Banner bool
+}
+
+func init() {
+ log.AddPackage(log.JSON, log.WarnLevel, nil)
+}
+
+// NewRWCoreFlags returns a new RWCore config
+func NewAdapterFlags() *AdapterFlags {
+ var adapterFlags = AdapterFlags{ // Default values
+ InstanceID: default_InstanceID,
+ KafkaAdapterHost: default_KafkaAdapterHost,
+ KafkaAdapterPort: default_KafkaAdapterPort,
+ KafkaClusterHost: default_KafkaClusterHost,
+ KafkaClusterPort: default_KafkaClusterPort,
+ KVStoreType: default_KVStoreType,
+ KVStoreTimeout: default_KVStoreTimeout,
+ KVStoreHost: default_KVStoreHost,
+ KVStorePort: default_KVStorePort,
+ Topic: default_Topic,
+ CoreTopic: default_CoreTopic,
+ LogLevel: default_LogLevel,
+ OnuNumber: default_OnuNumber,
+ Banner: default_Banner,
+ }
+ return &adapterFlags
+}
+
+// ParseCommandArguments parses the arguments when running read-write adaptercore service
+func (so *AdapterFlags) ParseCommandArguments() {
+
+ var help string
+
+ help = fmt.Sprintf("Kafka - Adapter messaging host")
+ flag.StringVar(&(so.KafkaAdapterHost), "kafka_adapter_host", default_KafkaAdapterHost, help)
+
+ help = fmt.Sprintf("Kafka - Adapter messaging port")
+ flag.IntVar(&(so.KafkaAdapterPort), "kafka_adapter_port", default_KafkaAdapterPort, help)
+
+ help = fmt.Sprintf("Kafka - Cluster messaging host")
+ flag.StringVar(&(so.KafkaClusterHost), "kafka_cluster_host", default_KafkaClusterHost, help)
+
+ help = fmt.Sprintf("Kafka - Cluster messaging port")
+ flag.IntVar(&(so.KafkaClusterPort), "kafka_cluster_port", default_KafkaClusterPort, help)
+
+ help = fmt.Sprintf("Simulated OLT topic")
+ flag.StringVar(&(so.Topic), "simulator_topic", default_Topic, help)
+
+ help = fmt.Sprintf("Core topic")
+ flag.StringVar(&(so.CoreTopic), "core_topic", default_CoreTopic, help)
+
+ help = fmt.Sprintf("KV store type")
+ flag.StringVar(&(so.KVStoreType), "kv_store_type", default_KVStoreType, help)
+
+ help = fmt.Sprintf("The default timeout when making a kv store request")
+ flag.IntVar(&(so.KVStoreTimeout), "kv_store_request_timeout", default_KVStoreTimeout, help)
+
+ help = fmt.Sprintf("KV store host")
+ flag.StringVar(&(so.KVStoreHost), "kv_store_host", default_KVStoreHost, help)
+
+ help = fmt.Sprintf("KV store port")
+ flag.IntVar(&(so.KVStorePort), "kv_store_port", default_KVStorePort, help)
+
+ help = fmt.Sprintf("Log level")
+ flag.IntVar(&(so.LogLevel), "log_level", default_LogLevel, help)
+
+ help = fmt.Sprintf("Number of ONUs")
+ flag.IntVar(&(so.OnuNumber), "onu_number", default_OnuNumber, help)
+
+ help = fmt.Sprintf("Show startup banner log lines")
+ flag.BoolVar(&so.Banner, "banner", default_Banner, help)
+
+ flag.Parse()
+
+ containerName := getContainerInfo()
+ if len(containerName) > 0 {
+ so.InstanceID = containerName
+ }
+
+}
+
+func getContainerInfo() string {
+ return os.Getenv("HOSTNAME")
+}
diff --git a/adapters/simulated_olt/main.go b/adapters/simulated_olt/main.go
new file mode 100644
index 0000000..01234a9
--- /dev/null
+++ b/adapters/simulated_olt/main.go
@@ -0,0 +1,337 @@
+/*
+ * Copyright 2018-present Open Networking Foundation
+
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+
+ * http://www.apache.org/licenses/LICENSE-2.0
+
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package main
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "github.com/opencord/voltha-go/adapters"
+ com "github.com/opencord/voltha-go/adapters/common"
+ ac "github.com/opencord/voltha-go/adapters/simulated_olt/adaptercore"
+ "github.com/opencord/voltha-go/adapters/simulated_olt/config"
+ "github.com/opencord/voltha-go/common/log"
+ "github.com/opencord/voltha-go/db/kvstore"
+ "github.com/opencord/voltha-go/kafka"
+ ic "github.com/opencord/voltha-go/protos/inter_container"
+ "github.com/opencord/voltha-go/protos/voltha"
+ "os"
+ "os/signal"
+ "strconv"
+ "syscall"
+ "time"
+)
+
+type adapter struct {
+ instanceId string
+ config *config.AdapterFlags
+ iAdapter adapters.IAdapter
+ kafkaClient kafka.Client
+ kvClient kvstore.Client
+ kip *kafka.InterContainerProxy
+ coreProxy *com.CoreProxy
+ halted bool
+ exitChannel chan int
+ receiverChannels []<-chan *ic.InterContainerMessage
+}
+
+func init() {
+ log.AddPackage(log.JSON, log.DebugLevel, nil)
+}
+
+func newAdapter(cf *config.AdapterFlags) *adapter {
+ var a adapter
+ a.instanceId = cf.InstanceID
+ a.config = cf
+ a.halted = false
+ a.exitChannel = make(chan int, 1)
+ a.receiverChannels = make([]<-chan *ic.InterContainerMessage, 0)
+ return &a
+}
+
+func (a *adapter) start(ctx context.Context) {
+ log.Info("Starting Core Adapter components")
+ var err error
+
+ // Setup KV Client
+ log.Debugw("create-kv-client", log.Fields{"kvstore": a.config.KVStoreType})
+ if err := a.setKVClient(); err != nil {
+ log.Fatal("error-setting-kv-client")
+ }
+
+ // Setup Kafka Client
+ if a.kafkaClient, err = newKafkaClient("sarama", a.config.KafkaAdapterHost, a.config.KafkaAdapterPort); err != nil {
+ log.Fatal("Unsupported-common-client")
+ }
+
+ // Start the common InterContainer Proxy - retries indefinitely
+ if a.kip, err = a.startInterContainerProxy(-1); err != nil {
+ log.Fatal("error-starting-inter-container-proxy")
+ }
+
+ // Create the core proxy to handle requests to the Core
+ a.coreProxy = com.NewCoreProxy(a.kip, a.config.Topic, a.config.CoreTopic)
+
+ // Create the simulated OLT adapter
+ if a.iAdapter, err = a.startSimulatedOLT(ctx, a.kip, a.coreProxy, a.config.OnuNumber); err != nil {
+ log.Fatal("error-starting-inter-container-proxy")
+ }
+
+ // Register the core request handler
+ if err = a.setupRequestHandler(a.instanceId, a.iAdapter); err != nil {
+ log.Fatal("error-setting-core-request-handler")
+ }
+
+ // Register this adapter to the Core - retries indefinitely
+ if err = a.registerWithCore(-1); err != nil {
+ log.Fatal("error-registering-with-core")
+ }
+}
+
+func (rw *adapter) stop() {
+ // Stop leadership tracking
+ rw.halted = true
+
+ // send exit signal
+ rw.exitChannel <- 0
+
+ // Cleanup - applies only if we had a kvClient
+ if rw.kvClient != nil {
+ // Release all reservations
+ if err := rw.kvClient.ReleaseAllReservations(); err != nil {
+ log.Infow("fail-to-release-all-reservations", log.Fields{"error": err})
+ }
+ // Close the DB connection
+ rw.kvClient.Close()
+ }
+
+ // TODO: More cleanup
+}
+
+func newKVClient(storeType string, address string, timeout int) (kvstore.Client, error) {
+
+ log.Infow("kv-store-type", log.Fields{"store": storeType})
+ switch storeType {
+ case "consul":
+ return kvstore.NewConsulClient(address, timeout)
+ case "etcd":
+ return kvstore.NewEtcdClient(address, timeout)
+ }
+ return nil, errors.New("unsupported-kv-store")
+}
+
+func newKafkaClient(clientType string, host string, port int) (kafka.Client, error) {
+
+ log.Infow("common-client-type", log.Fields{"client": clientType})
+ switch clientType {
+ case "sarama":
+ return kafka.NewSaramaClient(
+ kafka.Host(host),
+ kafka.Port(port),
+ kafka.ProducerReturnOnErrors(true),
+ kafka.ProducerReturnOnSuccess(true),
+ kafka.ProducerMaxRetries(6),
+ kafka.ProducerRetryBackoff(time.Millisecond*30)), nil
+ }
+ return nil, errors.New("unsupported-client-type")
+}
+
+func (a *adapter) setKVClient() error {
+ addr := a.config.KVStoreHost + ":" + strconv.Itoa(a.config.KVStorePort)
+ client, err := newKVClient(a.config.KVStoreType, addr, a.config.KVStoreTimeout)
+ if err != nil {
+ a.kvClient = nil
+ log.Error(err)
+ return err
+ }
+ a.kvClient = client
+ return nil
+}
+
+func toString(value interface{}) (string, error) {
+ switch t := value.(type) {
+ case []byte:
+ return string(value.([]byte)), nil
+ case string:
+ return value.(string), nil
+ default:
+ return "", fmt.Errorf("unexpected-type-%T", t)
+ }
+}
+
+func (a *adapter) startInterContainerProxy(retries int) (*kafka.InterContainerProxy, error) {
+ log.Infow("starting-intercontainer-messaging-proxy", log.Fields{"host": a.config.KafkaAdapterHost,
+ "port": a.config.KafkaAdapterPort, "topic": a.config.Topic})
+ var err error
+ var kip *kafka.InterContainerProxy
+ if kip, err = kafka.NewInterContainerProxy(
+ kafka.InterContainerHost(a.config.KafkaAdapterHost),
+ kafka.InterContainerPort(a.config.KafkaAdapterPort),
+ kafka.MsgClient(a.kafkaClient),
+ kafka.DefaultTopic(&kafka.Topic{Name: a.config.Topic})); err != nil {
+ log.Errorw("fail-to-create-common-proxy", log.Fields{"error": err})
+ return nil, err
+ }
+ count := 0
+ for {
+ if err = kip.Start(); err != nil {
+ log.Warnw("error-starting-messaging-proxy", log.Fields{"error": err})
+ if retries == count {
+ return nil, err
+ }
+ count = +1
+ // Take a nap before retrying
+ time.Sleep(2 * time.Second)
+ } else {
+ break
+ }
+ }
+
+ log.Info("common-messaging-proxy-created")
+ return kip, nil
+}
+
+func (a *adapter) startSimulatedOLT(ctx context.Context, kip *kafka.InterContainerProxy, cp *com.CoreProxy, onuNumber int) (*ac.SimulatedOLT, error) {
+ log.Info("starting-simulated-olt")
+ var err error
+ sOLT := ac.NewSimulatedOLT(ctx, a.kip, cp, onuNumber)
+
+ if err = sOLT.Start(ctx); err != nil {
+ log.Fatalw("error-starting-messaging-proxy", log.Fields{"error": err})
+ return nil, err
+ }
+
+ log.Info("simulated-olt-started")
+ return sOLT, nil
+}
+
+func (a *adapter) setupRequestHandler(coreInstanceId string, iadapter adapters.IAdapter) error {
+ log.Info("setting-request-handler")
+ requestProxy := com.NewRequestHandlerProxy(coreInstanceId, iadapter)
+ if err := a.kip.SubscribeWithRequestHandlerInterface(kafka.Topic{Name: a.config.Topic}, requestProxy); err != nil {
+ log.Errorw("request-handler-setup-failed", log.Fields{"error": err})
+ return err
+
+ }
+ log.Info("request-handler-setup-done")
+ return nil
+}
+
+func (a *adapter) registerWithCore(retries int) error {
+ log.Info("registering-with-core")
+ adapterDescription := &voltha.Adapter{Id: "simulated_olt", Vendor: "simulation Enterprise Inc"}
+ types := []*voltha.DeviceType{{Id: "simulated_olt", Adapter: "simulated_olt"}}
+ deviceTypes := &voltha.DeviceTypes{Items: types}
+ count := 0
+ for {
+ if err := a.coreProxy.RegisterAdapter(nil, adapterDescription, deviceTypes); err != nil {
+ log.Warnw("registering-with-core-failed", log.Fields{"error": err})
+ if retries == count {
+ return err
+ }
+ count += 1
+ // Take a nap before retrying
+ time.Sleep(2 * time.Second)
+ } else {
+ break
+ }
+ }
+ log.Info("registered-with-core")
+ return nil
+}
+
+func waitForExit() int {
+ signalChannel := make(chan os.Signal, 1)
+ signal.Notify(signalChannel,
+ syscall.SIGHUP,
+ syscall.SIGINT,
+ syscall.SIGTERM,
+ syscall.SIGQUIT)
+
+ exitChannel := make(chan int)
+
+ go func() {
+ s := <-signalChannel
+ switch s {
+ case syscall.SIGHUP,
+ syscall.SIGINT,
+ syscall.SIGTERM,
+ syscall.SIGQUIT:
+ log.Infow("closing-signal-received", log.Fields{"signal": s})
+ exitChannel <- 0
+ default:
+ log.Infow("unexpected-signal-received", log.Fields{"signal": s})
+ exitChannel <- 1
+ }
+ }()
+
+ code := <-exitChannel
+ return code
+}
+
+func printBanner() {
+ fmt.Println(" ____ _ _ _ _ ___ _ _____ ")
+ fmt.Println("/ ___|(_)_ __ ___ _ _| | __ _| |_ ___ __| |/ _ \\| | |_ _| ")
+ fmt.Println("\\___ \\| | '_ ` _ \\| | | | |/ _` | __/ _ \\/ _` | | | | | | | ")
+ fmt.Println(" ___) | | | | | | | |_| | | (_| | || __/ (_| | |_| | |___| | ")
+ fmt.Println("|____/|_|_| |_| |_|\\__,_|_|\\__,_|\\__\\___|\\__,_|\\___/|_____|_| ")
+ fmt.Println(" ")
+}
+
+func main() {
+ start := time.Now()
+
+ cf := config.NewAdapterFlags()
+ cf.ParseCommandArguments()
+
+ //// Setup logging
+
+ //Setup default logger - applies for packages that do not have specific logger set
+ if _, err := log.SetDefaultLogger(log.JSON, cf.LogLevel, log.Fields{"instanceId": cf.InstanceID}); err != nil {
+ log.With(log.Fields{"error": err}).Fatal("Cannot setup logging")
+ }
+
+ // Update all loggers (provisionned via init) with a common field
+ if err := log.UpdateAllLoggers(log.Fields{"instanceId": cf.InstanceID}); err != nil {
+ log.With(log.Fields{"error": err}).Fatal("Cannot setup logging")
+ }
+
+ log.SetPackageLogLevel("github.com/opencord/voltha-go/adapters/common", log.DebugLevel)
+
+ defer log.CleanUp()
+
+ // Print banner if specified
+ if cf.Banner {
+ printBanner()
+ }
+
+ log.Infow("config", log.Fields{"config": *cf})
+
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ ad := newAdapter(cf)
+ go ad.start(ctx)
+
+ code := waitForExit()
+ log.Infow("received-a-closing-signal", log.Fields{"code": code})
+
+ // Cleanup before leaving
+ ad.stop()
+
+ elapsed := time.Since(start)
+ log.Infow("run-time", log.Fields{"instanceId": ad.config.InstanceID, "time": elapsed / time.Second})
+}