core/olt_manager.go - openolt-scale-tester - Gitiles

 /*
  * 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 core

 import (
 	"context"
 	"encoding/hex"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"github.com/golang/protobuf/jsonpb"
 	ponrmgr "github.com/opencord/voltha-lib-go/v7/pkg/ponresourcemanager"
 	"github.com/opencord/voltha-protos/v5/go/tech_profile"
 	"io"
 	"os"
 	"strconv"
 	"sync"
 	"syscall"
 	"time"

 	"github.com/cenkalti/backoff/v3"
 	"github.com/opencord/openolt-scale-tester/config"
 	"github.com/opencord/voltha-lib-go/v7/pkg/db/kvstore"
 	"github.com/opencord/voltha-lib-go/v7/pkg/log"
 	oop "github.com/opencord/voltha-protos/v5/go/openolt"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 )

 const (
 	ReasonOk          = "OK"
 	TechProfileKVPath = "service/voltha/technology_profiles/%s/%d" // service/voltha/technology_profiles/xgspon/<tech_profile_tableID>
 )

 type OnuDeviceKey struct {
 	onuID    uint32
 	ponInfID uint32
 }

 type OpenOltManager struct {
 	ipPort             string
 	deviceInfo         *oop.DeviceInfo
 	OnuDeviceMap       map[OnuDeviceKey]*OnuDevice `json:"onuDeviceMap"`
 	clientConn         *grpc.ClientConn
 	openOltClient      oop.OpenoltClient
 	testConfig         *config.OpenOltScaleTesterConfig
 	rsrMgr             []*OpenOltResourceMgr
 	lockRsrAlloc       sync.RWMutex
 	lockOpenOltManager sync.RWMutex
 }

 func NewOpenOltManager(ipPort string) *OpenOltManager {
 	logger.Infow(nil, "initialized openolt manager with ipPort", log.Fields{"ipPort": ipPort})
 	return &OpenOltManager{
 		ipPort:             ipPort,
 		OnuDeviceMap:       make(map[OnuDeviceKey]*OnuDevice),
 		lockRsrAlloc:       sync.RWMutex{},
 		lockOpenOltManager: sync.RWMutex{},
 	}
 }

 func (om *OpenOltManager) readAndLoadTPsToEtcd() {
 	var err error
 	// Verify that etcd is up before starting the application.
 	etcdIpPort := "http://" + om.testConfig.KVStoreHost + ":" + strconv.Itoa(om.testConfig.KVStorePort)
 	client, err := kvstore.NewEtcdClient(context.Background(), etcdIpPort, 5*time.Second, log.FatalLevel)
 	if err != nil || client == nil {
 		logger.Fatal(nil, "error-initializing-etcd-client")
 		return
 	}

 	// Load TPs to etcd for each of the specified tech-profiles
 	for _, tpID := range om.testConfig.TpIDList {
 		// Below should translate to something like "/app/ATT-64.json"
 		// The TP file should exist.
 		tpFilePath := "/app/" + om.testConfig.WorkflowName + "-" + strconv.Itoa(tpID) + ".json"
 		// Open our jsonFile
 		jsonFile, err := os.Open(tpFilePath)
 		// if we os.Open returns an error then handle it
 		if err != nil {
 			logger.Fatalw(nil, "could-not-find-tech-profile", log.Fields{"err": err, "tpFile": tpFilePath})
 		}
 		logger.Debugw(nil, "tp-file-opened-successfully", log.Fields{"tpFile": tpFilePath})

 		/*
 			// read our opened json file as a byte array.
 			if byteValue, err = ioutil.ReadAll(jsonFile); err != nil {
 				logger.Fatalw(nil, "could-not-read-tp-file", log.Fields{"err": err, "tpFile": tpFilePath})
 			}

 		*/

 		var tp tech_profile.TechProfile

 		if err = jsonpb.Unmarshal(jsonFile, &tp); err != nil {
 			logger.Fatalw(nil, "could-not-unmarshal-tp", log.Fields{"err": err, "tpFile": tpFilePath})
 		} else {
 			logger.Infow(nil, "tp-read-from-file", log.Fields{"tp": tp, "tpFile": tpFilePath})
 		}
 		kvPath := fmt.Sprintf(TechProfileKVPath, om.deviceInfo.Technology, tpID)
 		tpJson, err := json.Marshal(tp)
 		err = client.Put(context.Background(), kvPath, tpJson)
 		if err != nil {
 			logger.Fatalw(nil, "tp-put-to-etcd-failed", log.Fields{"tpPath": kvPath, "err": err})
 		}
 		// verify the PUT succeeded.
 		kvResult, err := client.Get(context.Background(), kvPath)
 		if kvResult == nil {
 			logger.Fatal(nil, "tp-not-found-on-kv-after-load", log.Fields{"key": kvPath, "err": err})
 		} else {
 			var KvTp tech_profile.TechProfile
 			var resPtr = &KvTp
 			if value, err := kvstore.ToByte(kvResult.Value); err == nil {
 				if err = json.Unmarshal(value, resPtr); err != nil {
 					logger.Fatal(nil, "error-unmarshal-kv-result", log.Fields{"err": err, "key": kvPath, "value": value})
 				} else {
 					logger.Infow(nil, "verified-ok-that-tp-load-was-good", log.Fields{"tpID": tpID, "kvPath": kvPath})
 					_ = jsonFile.Close()
 					continue
 				}
 			}
 		}
 	}
 }

 func (om *OpenOltManager) Start(testConfig *config.OpenOltScaleTesterConfig) error {
 	var err error
 	om.testConfig = testConfig

 	// Establish gRPC connection with the device
 	if om.clientConn, err = grpc.Dial(om.ipPort, grpc.WithInsecure(), grpc.WithBlock()); err != nil {
 		logger.Errorw(nil, "Failed to dial device", log.Fields{"ipPort": om.ipPort, "err": err})
 		return err
 	}
 	om.openOltClient = oop.NewOpenoltClient(om.clientConn)

 	// Populate Device Info
 	if deviceInfo, err := om.populateDeviceInfo(); err != nil {
 		logger.Error(nil, "error fetching device info", log.Fields{"err": err, "deviceInfo": deviceInfo})
 		return err
 	}

 	// Read and load TPs to etcd.
 	om.readAndLoadTPsToEtcd()

 	logger.Info(nil, "etcd-up-and-running--tp-loaded-successfully")
 	om.rsrMgr = make([]*OpenOltResourceMgr, om.deviceInfo.PonPorts)
 	for ponIdx := range om.rsrMgr {
 		if om.rsrMgr[ponIdx] = NewResourceMgr("ABCD", om.testConfig.KVStoreHost+":"+strconv.Itoa(om.testConfig.KVStorePort),
 			"etcd", "openolt", uint32(ponIdx), om.deviceInfo); om.rsrMgr == nil {
 			logger.Error(nil, "Error while instantiating resource manager")
 			return errors.New("instantiating resource manager failed")
 		}
 	}

 	// Start reading indications
 	go om.readIndications()

 	// Provision OLT NNI Trap flows as needed by the Workflow
 	if err = ProvisionNniTrapFlow(om.openOltClient, om.testConfig, om.rsrMgr[0]); err != nil {
 		logger.Error(nil, "failed-to-add-nni-trap-flow", log.Fields{"err": err})
 	}

 	// Provision ONUs one by one
 	go om.provisionONUs()

 	return nil

 }

 func (om *OpenOltManager) populateDeviceInfo() (*oop.DeviceInfo, error) {
 	var err error

 	if om.deviceInfo, err = om.openOltClient.GetDeviceInfo(context.Background(), new(oop.Empty)); err != nil {
 		logger.Errorw(nil, "Failed to fetch device info", log.Fields{"err": err})
 		return nil, err
 	}

 	if om.deviceInfo == nil {
 		logger.Errorw(nil, "Device info is nil", log.Fields{})
 		return nil, errors.New("failed to get device info from OLT")
 	}

 	logger.Debugw(nil, "Fetched device info", log.Fields{"deviceInfo": om.deviceInfo})

 	return om.deviceInfo, nil
 }

 func (om *OpenOltManager) provisionONUs() {
 	var numOfONUsPerPon uint
 	var i, j, k uint32
 	var onuWg sync.WaitGroup

 	defer func() {
 		// Stop the process once the job is done
 		_ = syscall.Kill(syscall.Getpid(), syscall.SIGINT)
 	}()

 	// If the number of ONUs to provision is not a power of 2, stop execution
 	// This is needed for ensure even distribution of ONUs across all PONs
 	if !isPowerOfTwo(om.testConfig.NumOfOnu) {
 		logger.Errorw(nil, "num-of-onus-to-provision-is-not-a-power-of-2", log.Fields{"numOfOnus": om.testConfig.NumOfOnu})
 		return
 	}

 	// Number of ONUs to provision should not be less than the number of PON ports.
 	// We need at least one ONU per PON
 	if om.testConfig.NumOfOnu < uint(om.deviceInfo.PonPorts) {
 		logger.Errorw(nil, "num-of-onu-is-less-than-num-of-pon-port", log.Fields{"numOfOnus": om.testConfig.NumOfOnu, "numOfPon": om.deviceInfo.PonPorts})
 		return
 	}

 	numOfONUsPerPon = om.testConfig.NumOfOnu / uint(om.deviceInfo.PonPorts)
 	totalOnusToProvision := numOfONUsPerPon * uint(om.deviceInfo.PonPorts)
 	logger.Infow(nil, "***** all-onu-provision-started ******",
 		log.Fields{"totalNumOnus": totalOnusToProvision,
 			"numOfOnusPerPon": numOfONUsPerPon,
 			"numOfPons":       om.deviceInfo.PonPorts})

 	// These are the number of ONUs that will be provisioned per PON port per batch.
 	// Such number of ONUs will be chosen across all PON ports per batch
 	var onusPerIterationPerPonPort uint32 = 4

 	// If the total number of ONUs per PON is lesser than the default ONU to provision per pon port per batch
 	// then keep halving the ONU to provision per pon port per batch until we reach an acceptable number
 	// Note: the least possible value for onusPerIterationPerPonPort is 1
 	for uint32(numOfONUsPerPon) < onusPerIterationPerPonPort {
 		onusPerIterationPerPonPort /= 2
 	}

 	startTime := time.Now()
 	// Start provisioning the ONUs
 	for i = 0; i < uint32(numOfONUsPerPon)/onusPerIterationPerPonPort; i++ {
 		for j = 0; j < om.deviceInfo.PonPorts; j++ {
 			for k = 0; k < onusPerIterationPerPonPort; k++ {
 				om.lockRsrAlloc.Lock()
 				sn := GenerateNextONUSerialNumber()
 				om.lockRsrAlloc.Unlock()
 				logger.Debugw(nil, "provisioning onu", log.Fields{"onuID": j, "ponPort": i, "serialNum": sn})
 				ctx := context.Background()
 				newCtx := context.WithValue(ctx, "ponIf", i)
 				onuIDSl, err := om.rsrMgr[i].TechprofileRef.GetResourceID(newCtx, j, ponrmgr.ONU_ID, 1)
 				if err != nil {
 					logger.Errorw(nil, "error getting onu id", log.Fields{"err": err})
 					continue
 				}
 				logger.Infow(nil, "onu-provision-started-from-olt-manager", log.Fields{"onuId": onuIDSl[0], "ponIntf": i})

 				onuWg.Add(1)
 				go om.activateONU(j, onuIDSl[0], sn, om.stringifySerialNumber(sn), &onuWg)
 			}
 		}
 		// Wait for the group of ONUs to complete processing before going to next batch of ONUs
 		onuWg.Wait()
 	}
 	endTime := time.Now()
 	logger.Info(nil, "******** all-onu-provisioning-completed *******")
 	totalTime := endTime.Sub(startTime)
 	out := time.Time{}.Add(totalTime)
 	logger.Infof(nil, "****** Total Time to provision all the ONUs is => %s", out.Format("15:04:05"))

 	// TODO: We need to dump the results at the end. But below json marshall does not work. We will need custom Marshal function.
 	/*
 		e, err := json.Marshal(om)
 		if err != nil {
 			fmt.Println(err)
 			return
 		}
 		fmt.Println(string(e))
 	*/
 }

 func (om *OpenOltManager) activateONU(intfID uint32, onuID uint32, serialNum *oop.SerialNumber, serialNumber string, onuWg *sync.WaitGroup) {
 	logger.Debugw(nil, "activate-onu", log.Fields{"intfID": intfID, "onuID": onuID, "serialNum": serialNum, "serialNumber": serialNumber})
 	// TODO: need resource manager
 	var pir uint32 = 1000000
 	var onuDevice = OnuDevice{
 		SerialNum:     serialNumber,
 		OnuID:         onuID,
 		PonIntf:       intfID,
 		openOltClient: om.openOltClient,
 		testConfig:    om.testConfig,
 		rsrMgr:        om.rsrMgr[intfID],
 		onuWg:         onuWg,
 	}
 	var err error
 	onuDeviceKey := OnuDeviceKey{onuID: onuID, ponInfID: intfID}
 	Onu := oop.Onu{IntfId: intfID, OnuId: onuID, SerialNumber: serialNum, Pir: pir}
 	now := time.Now()
 	nanos := now.UnixNano()
 	milliStart := nanos / 1000000
 	onuDevice.OnuProvisionStartTime = time.Unix(0, nanos)
 	if _, err = om.openOltClient.ActivateOnu(context.Background(), &Onu); err != nil {
 		st, _ := status.FromError(err)
 		if st.Code() == codes.AlreadyExists {
 			logger.Debug(nil, "ONU activation is in progress", log.Fields{"SerialNumber": serialNumber})
 		} else {
 			nanos = now.UnixNano()
 			milliEnd := nanos / 1000000
 			onuDevice.OnuProvisionEndTime = time.Unix(0, nanos)
 			onuDevice.OnuProvisionDurationInMs = milliEnd - milliStart
 			logger.Errorw(nil, "activate-onu-failed", log.Fields{"Onu": Onu, "err ": err})
 			onuDevice.Reason = err.Error()
 		}
 	} else {
 		nanos = now.UnixNano()
 		milliEnd := nanos / 1000000
 		onuDevice.OnuProvisionEndTime = time.Unix(0, nanos)
 		onuDevice.OnuProvisionDurationInMs = milliEnd - milliStart
 		onuDevice.Reason = ReasonOk
 		logger.Infow(nil, "activated-onu", log.Fields{"SerialNumber": serialNumber})
 	}

 	om.lockOpenOltManager.Lock()
 	om.OnuDeviceMap[onuDeviceKey] = &onuDevice
 	om.lockOpenOltManager.Unlock()

 	// If ONU activation was success provision the ONU
 	if err == nil {
 		om.lockOpenOltManager.RLock()
 		go om.OnuDeviceMap[onuDeviceKey].Start()
 		om.lockOpenOltManager.RUnlock()

 	}
 }

 func (om *OpenOltManager) stringifySerialNumber(serialNum *oop.SerialNumber) string {
 	if serialNum != nil {
 		return string(serialNum.VendorId) + om.stringifyVendorSpecific(serialNum.VendorSpecific)
 	}
 	return ""
 }

 func (om *OpenOltManager) stringifyVendorSpecific(vendorSpecific []byte) string {
 	tmp := fmt.Sprintf("%x", (uint32(vendorSpecific[0])>>4)&0x0f) +
 		fmt.Sprintf("%x", uint32(vendorSpecific[0]&0x0f)) +
 		fmt.Sprintf("%x", (uint32(vendorSpecific[1])>>4)&0x0f) +
 		fmt.Sprintf("%x", (uint32(vendorSpecific[1]))&0x0f) +
 		fmt.Sprintf("%x", (uint32(vendorSpecific[2])>>4)&0x0f) +
 		fmt.Sprintf("%x", (uint32(vendorSpecific[2]))&0x0f) +
 		fmt.Sprintf("%x", (uint32(vendorSpecific[3])>>4)&0x0f) +
 		fmt.Sprintf("%x", (uint32(vendorSpecific[3]))&0x0f)
 	return tmp
 }

 // readIndications to read the indications from the OLT device
 func (om *OpenOltManager) readIndications() {
 	defer logger.Errorw(nil, "Indications ended", log.Fields{})
 	indications, err := om.openOltClient.EnableIndication(context.Background(), new(oop.Empty))
 	if err != nil {
 		logger.Errorw(nil, "Failed to read indications", log.Fields{"err": err})
 		return
 	}
 	if indications == nil {
 		logger.Errorw(nil, "Indications is nil", log.Fields{})
 		return
 	}

 	// Create an exponential backoff around re-enabling indications. The
 	// maximum elapsed time for the back off is set to 0 so that we will
 	// continue to retry. The max interval defaults to 1m, but is set
 	// here for code clarity
 	indicationBackoff := backoff.NewExponentialBackOff()
 	indicationBackoff.MaxElapsedTime = 0
 	indicationBackoff.MaxInterval = 1 * time.Minute
 	for {
 		indication, err := indications.Recv()
 		if err == io.EOF {
 			logger.Infow(nil, "EOF for  indications", log.Fields{"err": err})
 			// Use an exponential back off to prevent getting into a tight loop
 			duration := indicationBackoff.NextBackOff()
 			if duration == backoff.Stop {
 				// If we reach a maximum then warn and reset the backoff
 				// timer and keep attempting.
 				logger.Warnw(nil, "Maximum indication backoff reached, resetting backoff timer",
 					log.Fields{"max_indication_backoff": indicationBackoff.MaxElapsedTime})
 				indicationBackoff.Reset()
 			}
 			time.Sleep(indicationBackoff.NextBackOff())
 			indications, err = om.openOltClient.EnableIndication(context.Background(), new(oop.Empty))
 			if err != nil {
 				logger.Errorw(nil, "Failed to read indications", log.Fields{"err": err})
 				return
 			}
 			continue
 		}
 		if err != nil {
 			logger.Infow(nil, "Failed to read from indications", log.Fields{"err": err})
 			break
 		}
 		// Reset backoff if we have a successful receive
 		indicationBackoff.Reset()
 		om.handleIndication(indication)

 	}
 }

 func (om *OpenOltManager) handleIndication(indication *oop.Indication) {
 	switch indication.Data.(type) {
 	case *oop.Indication_OltInd:
 		logger.Info(nil, "received olt indication")
 	case *oop.Indication_IntfInd:
 		intfInd := indication.GetIntfInd()
 		logger.Infow(nil, "Received interface indication ", log.Fields{"InterfaceInd": intfInd})
 	case *oop.Indication_IntfOperInd:
 		intfOperInd := indication.GetIntfOperInd()
 		if intfOperInd.GetType() == "nni" {
 			logger.Info(nil, "received interface oper indication for nni port")
 		} else if intfOperInd.GetType() == "pon" {
 			logger.Info(nil, "received interface oper indication for pon port")
 		}
 		/*
 			case *oop.Indication_OnuDiscInd:
 				onuDiscInd := indication.GetOnuDiscInd()
 				logger.Infow(nil, "Received Onu discovery indication ", log.Fields{"OnuDiscInd": onuDiscInd})
 		*/
 	case *oop.Indication_OnuInd:
 		onuInd := indication.GetOnuInd()
 		logger.Infow(nil, "Received Onu indication ", log.Fields{"OnuInd": onuInd})
 	case *oop.Indication_OmciInd:
 		omciInd := indication.GetOmciInd()
 		logger.Debugw(nil, "Received Omci indication ", log.Fields{"IntfId": omciInd.IntfId, "OnuId": omciInd.OnuId, "pkt": hex.EncodeToString(omciInd.Pkt)})
 	case *oop.Indication_PktInd:
 		pktInd := indication.GetPktInd()
 		logger.Infow(nil, "Received packet indication ", log.Fields{"PktInd": pktInd})
 		/*
 				case *oop.Indication_PortStats:
 				portStats := indication.GetPortStats()
 				logger.Infow(nil, "Received port stats", log.Fields{"portStats": portStats})
 			case *oop.Indication_FlowStats:
 				flowStats := indication.GetFlowStats()
 				logger.Infow(nil, "Received flow stats", log.Fields{"FlowStats": flowStats})
 		*/
 	case *oop.Indication_AlarmInd:
 		alarmInd := indication.GetAlarmInd()
 		logger.Infow(nil, "Received alarm indication ", log.Fields{"AlarmInd": alarmInd})
 	}
 }

 func isPowerOfTwo(numOfOnus uint) bool {
 	return (numOfOnus & (numOfOnus - 1)) == 0
 }
	/*
	* Copyright 2018-present Open Networking Foundation

	* Licensed under the Apache License, Version 2.0 (the "License");
	* you may not use this file except in compliance with the License.
	* You may obtain a copy of the License at

	* http://www.apache.org/licenses/LICENSE-2.0

	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/

	package core

	import (
	"context"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/golang/protobuf/jsonpb"
	ponrmgr "github.com/opencord/voltha-lib-go/v7/pkg/ponresourcemanager"
	"github.com/opencord/voltha-protos/v5/go/tech_profile"
	"io"
	"os"
	"strconv"
	"sync"
	"syscall"
	"time"

	"github.com/cenkalti/backoff/v3"
	"github.com/opencord/openolt-scale-tester/config"
	"github.com/opencord/voltha-lib-go/v7/pkg/db/kvstore"
	"github.com/opencord/voltha-lib-go/v7/pkg/log"
	oop "github.com/opencord/voltha-protos/v5/go/openolt"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	)

	const (
	ReasonOk = "OK"
	TechProfileKVPath = "service/voltha/technology_profiles/%s/%d" // service/voltha/technology_profiles/xgspon/<tech_profile_tableID>
	)

	type OnuDeviceKey struct {
	onuID uint32
	ponInfID uint32
	}

	type OpenOltManager struct {
	ipPort string
	deviceInfo *oop.DeviceInfo
	OnuDeviceMap map[OnuDeviceKey]*OnuDevice `json:"onuDeviceMap"`
	clientConn *grpc.ClientConn
	openOltClient oop.OpenoltClient
	testConfig *config.OpenOltScaleTesterConfig
	rsrMgr []*OpenOltResourceMgr
	lockRsrAlloc sync.RWMutex
	lockOpenOltManager sync.RWMutex
	}

	func NewOpenOltManager(ipPort string) *OpenOltManager {
	logger.Infow(nil, "initialized openolt manager with ipPort", log.Fields{"ipPort": ipPort})
	return &OpenOltManager{
	ipPort: ipPort,
	OnuDeviceMap: make(map[OnuDeviceKey]*OnuDevice),
	lockRsrAlloc: sync.RWMutex{},
	lockOpenOltManager: sync.RWMutex{},
	}
	}

	func (om *OpenOltManager) readAndLoadTPsToEtcd() {
	var err error
	// Verify that etcd is up before starting the application.
	etcdIpPort := "http://" + om.testConfig.KVStoreHost + ":" + strconv.Itoa(om.testConfig.KVStorePort)
	client, err := kvstore.NewEtcdClient(context.Background(), etcdIpPort, 5*time.Second, log.FatalLevel)
	if err != nil \|\| client == nil {
	logger.Fatal(nil, "error-initializing-etcd-client")
	return
	}

	// Load TPs to etcd for each of the specified tech-profiles
	for _, tpID := range om.testConfig.TpIDList {
	// Below should translate to something like "/app/ATT-64.json"
	// The TP file should exist.
	tpFilePath := "/app/" + om.testConfig.WorkflowName + "-" + strconv.Itoa(tpID) + ".json"
	// Open our jsonFile
	jsonFile, err := os.Open(tpFilePath)
	// if we os.Open returns an error then handle it
	if err != nil {
	logger.Fatalw(nil, "could-not-find-tech-profile", log.Fields{"err": err, "tpFile": tpFilePath})
	}
	logger.Debugw(nil, "tp-file-opened-successfully", log.Fields{"tpFile": tpFilePath})

	/*
	// read our opened json file as a byte array.
	if byteValue, err = ioutil.ReadAll(jsonFile); err != nil {
	logger.Fatalw(nil, "could-not-read-tp-file", log.Fields{"err": err, "tpFile": tpFilePath})
	}

	*/

	var tp tech_profile.TechProfile

	if err = jsonpb.Unmarshal(jsonFile, &tp); err != nil {
	logger.Fatalw(nil, "could-not-unmarshal-tp", log.Fields{"err": err, "tpFile": tpFilePath})
	} else {
	logger.Infow(nil, "tp-read-from-file", log.Fields{"tp": tp, "tpFile": tpFilePath})
	}
	kvPath := fmt.Sprintf(TechProfileKVPath, om.deviceInfo.Technology, tpID)
	tpJson, err := json.Marshal(tp)
	err = client.Put(context.Background(), kvPath, tpJson)
	if err != nil {
	logger.Fatalw(nil, "tp-put-to-etcd-failed", log.Fields{"tpPath": kvPath, "err": err})
	}
	// verify the PUT succeeded.
	kvResult, err := client.Get(context.Background(), kvPath)
	if kvResult == nil {
	logger.Fatal(nil, "tp-not-found-on-kv-after-load", log.Fields{"key": kvPath, "err": err})
	} else {
	var KvTp tech_profile.TechProfile
	var resPtr = &KvTp
	if value, err := kvstore.ToByte(kvResult.Value); err == nil {
	if err = json.Unmarshal(value, resPtr); err != nil {
	logger.Fatal(nil, "error-unmarshal-kv-result", log.Fields{"err": err, "key": kvPath, "value": value})
	} else {
	logger.Infow(nil, "verified-ok-that-tp-load-was-good", log.Fields{"tpID": tpID, "kvPath": kvPath})
	_ = jsonFile.Close()
	continue
	}
	}
	}
	}
	}

	func (om OpenOltManager) Start(testConfig config.OpenOltScaleTesterConfig) error {
	var err error
	om.testConfig = testConfig

	// Establish gRPC connection with the device
	if om.clientConn, err = grpc.Dial(om.ipPort, grpc.WithInsecure(), grpc.WithBlock()); err != nil {
	logger.Errorw(nil, "Failed to dial device", log.Fields{"ipPort": om.ipPort, "err": err})
	return err
	}
	om.openOltClient = oop.NewOpenoltClient(om.clientConn)

	// Populate Device Info
	if deviceInfo, err := om.populateDeviceInfo(); err != nil {
	logger.Error(nil, "error fetching device info", log.Fields{"err": err, "deviceInfo": deviceInfo})
	return err
	}

	// Read and load TPs to etcd.
	om.readAndLoadTPsToEtcd()

	logger.Info(nil, "etcd-up-and-running--tp-loaded-successfully")
	om.rsrMgr = make([]*OpenOltResourceMgr, om.deviceInfo.PonPorts)
	for ponIdx := range om.rsrMgr {
	if om.rsrMgr[ponIdx] = NewResourceMgr("ABCD", om.testConfig.KVStoreHost+":"+strconv.Itoa(om.testConfig.KVStorePort),
	"etcd", "openolt", uint32(ponIdx), om.deviceInfo); om.rsrMgr == nil {
	logger.Error(nil, "Error while instantiating resource manager")
	return errors.New("instantiating resource manager failed")
	}
	}

	// Start reading indications
	go om.readIndications()

	// Provision OLT NNI Trap flows as needed by the Workflow
	if err = ProvisionNniTrapFlow(om.openOltClient, om.testConfig, om.rsrMgr[0]); err != nil {
	logger.Error(nil, "failed-to-add-nni-trap-flow", log.Fields{"err": err})
	}

	// Provision ONUs one by one
	go om.provisionONUs()

	return nil

	}

	func (om OpenOltManager) populateDeviceInfo() (oop.DeviceInfo, error) {
	var err error

	if om.deviceInfo, err = om.openOltClient.GetDeviceInfo(context.Background(), new(oop.Empty)); err != nil {
	logger.Errorw(nil, "Failed to fetch device info", log.Fields{"err": err})
	return nil, err
	}

	if om.deviceInfo == nil {
	logger.Errorw(nil, "Device info is nil", log.Fields{})
	return nil, errors.New("failed to get device info from OLT")
	}

	logger.Debugw(nil, "Fetched device info", log.Fields{"deviceInfo": om.deviceInfo})

	return om.deviceInfo, nil
	}

	func (om *OpenOltManager) provisionONUs() {
	var numOfONUsPerPon uint
	var i, j, k uint32
	var onuWg sync.WaitGroup

	defer func() {
	// Stop the process once the job is done
	_ = syscall.Kill(syscall.Getpid(), syscall.SIGINT)
	}()

	// If the number of ONUs to provision is not a power of 2, stop execution
	// This is needed for ensure even distribution of ONUs across all PONs
	if !isPowerOfTwo(om.testConfig.NumOfOnu) {
	logger.Errorw(nil, "num-of-onus-to-provision-is-not-a-power-of-2", log.Fields{"numOfOnus": om.testConfig.NumOfOnu})
	return
	}

	// Number of ONUs to provision should not be less than the number of PON ports.
	// We need at least one ONU per PON
	if om.testConfig.NumOfOnu < uint(om.deviceInfo.PonPorts) {
	logger.Errorw(nil, "num-of-onu-is-less-than-num-of-pon-port", log.Fields{"numOfOnus": om.testConfig.NumOfOnu, "numOfPon": om.deviceInfo.PonPorts})
	return
	}

	numOfONUsPerPon = om.testConfig.NumOfOnu / uint(om.deviceInfo.PonPorts)
	totalOnusToProvision := numOfONUsPerPon * uint(om.deviceInfo.PonPorts)
	logger.Infow(nil, "*** all-onu-provision-started ****",
	log.Fields{"totalNumOnus": totalOnusToProvision,
	"numOfOnusPerPon": numOfONUsPerPon,
	"numOfPons": om.deviceInfo.PonPorts})

	// These are the number of ONUs that will be provisioned per PON port per batch.
	// Such number of ONUs will be chosen across all PON ports per batch
	var onusPerIterationPerPonPort uint32 = 4

	// If the total number of ONUs per PON is lesser than the default ONU to provision per pon port per batch
	// then keep halving the ONU to provision per pon port per batch until we reach an acceptable number
	// Note: the least possible value for onusPerIterationPerPonPort is 1
	for uint32(numOfONUsPerPon) < onusPerIterationPerPonPort {
	onusPerIterationPerPonPort /= 2
	}

	startTime := time.Now()
	// Start provisioning the ONUs
	for i = 0; i < uint32(numOfONUsPerPon)/onusPerIterationPerPonPort; i++ {
	for j = 0; j < om.deviceInfo.PonPorts; j++ {
	for k = 0; k < onusPerIterationPerPonPort; k++ {
	om.lockRsrAlloc.Lock()
	sn := GenerateNextONUSerialNumber()
	om.lockRsrAlloc.Unlock()
	logger.Debugw(nil, "provisioning onu", log.Fields{"onuID": j, "ponPort": i, "serialNum": sn})
	ctx := context.Background()
	newCtx := context.WithValue(ctx, "ponIf", i)
	onuIDSl, err := om.rsrMgr[i].TechprofileRef.GetResourceID(newCtx, j, ponrmgr.ONU_ID, 1)
	if err != nil {
	logger.Errorw(nil, "error getting onu id", log.Fields{"err": err})
	continue
	}
	logger.Infow(nil, "onu-provision-started-from-olt-manager", log.Fields{"onuId": onuIDSl[0], "ponIntf": i})

	onuWg.Add(1)
	go om.activateONU(j, onuIDSl[0], sn, om.stringifySerialNumber(sn), &onuWg)
	}
	}
	// Wait for the group of ONUs to complete processing before going to next batch of ONUs
	onuWg.Wait()
	}
	endTime := time.Now()
	logger.Info(nil, "****** all-onu-provisioning-completed *****")
	totalTime := endTime.Sub(startTime)
	out := time.Time{}.Add(totalTime)
	logger.Infof(nil, "****** Total Time to provision all the ONUs is => %s", out.Format("15:04:05"))

	// TODO: We need to dump the results at the end. But below json marshall does not work. We will need custom Marshal function.
	/*
	e, err := json.Marshal(om)
	if err != nil {
	fmt.Println(err)
	return
	}
	fmt.Println(string(e))
	*/
	}

	func (om OpenOltManager) activateONU(intfID uint32, onuID uint32, serialNum oop.SerialNumber, serialNumber string, onuWg *sync.WaitGroup) {
	logger.Debugw(nil, "activate-onu", log.Fields{"intfID": intfID, "onuID": onuID, "serialNum": serialNum, "serialNumber": serialNumber})
	// TODO: need resource manager
	var pir uint32 = 1000000
	var onuDevice = OnuDevice{
	SerialNum: serialNumber,
	OnuID: onuID,
	PonIntf: intfID,
	openOltClient: om.openOltClient,
	testConfig: om.testConfig,
	rsrMgr: om.rsrMgr[intfID],
	onuWg: onuWg,
	}
	var err error
	onuDeviceKey := OnuDeviceKey{onuID: onuID, ponInfID: intfID}
	Onu := oop.Onu{IntfId: intfID, OnuId: onuID, SerialNumber: serialNum, Pir: pir}
	now := time.Now()
	nanos := now.UnixNano()
	milliStart := nanos / 1000000
	onuDevice.OnuProvisionStartTime = time.Unix(0, nanos)
	if _, err = om.openOltClient.ActivateOnu(context.Background(), &Onu); err != nil {
	st, _ := status.FromError(err)
	if st.Code() == codes.AlreadyExists {
	logger.Debug(nil, "ONU activation is in progress", log.Fields{"SerialNumber": serialNumber})
	} else {
	nanos = now.UnixNano()
	milliEnd := nanos / 1000000
	onuDevice.OnuProvisionEndTime = time.Unix(0, nanos)
	onuDevice.OnuProvisionDurationInMs = milliEnd - milliStart
	logger.Errorw(nil, "activate-onu-failed", log.Fields{"Onu": Onu, "err ": err})
	onuDevice.Reason = err.Error()
	}
	} else {
	nanos = now.UnixNano()
	milliEnd := nanos / 1000000
	onuDevice.OnuProvisionEndTime = time.Unix(0, nanos)
	onuDevice.OnuProvisionDurationInMs = milliEnd - milliStart
	onuDevice.Reason = ReasonOk
	logger.Infow(nil, "activated-onu", log.Fields{"SerialNumber": serialNumber})
	}

	om.lockOpenOltManager.Lock()
	om.OnuDeviceMap[onuDeviceKey] = &onuDevice
	om.lockOpenOltManager.Unlock()

	// If ONU activation was success provision the ONU
	if err == nil {
	om.lockOpenOltManager.RLock()
	go om.OnuDeviceMap[onuDeviceKey].Start()
	om.lockOpenOltManager.RUnlock()

	}
	}

	func (om OpenOltManager) stringifySerialNumber(serialNum oop.SerialNumber) string {
	if serialNum != nil {
	return string(serialNum.VendorId) + om.stringifyVendorSpecific(serialNum.VendorSpecific)
	}
	return ""
	}

	func (om *OpenOltManager) stringifyVendorSpecific(vendorSpecific []byte) string {
	tmp := fmt.Sprintf("%x", (uint32(vendorSpecific[0])>>4)&0x0f) +
	fmt.Sprintf("%x", uint32(vendorSpecific[0]&0x0f)) +
	fmt.Sprintf("%x", (uint32(vendorSpecific[1])>>4)&0x0f) +
	fmt.Sprintf("%x", (uint32(vendorSpecific[1]))&0x0f) +
	fmt.Sprintf("%x", (uint32(vendorSpecific[2])>>4)&0x0f) +
	fmt.Sprintf("%x", (uint32(vendorSpecific[2]))&0x0f) +
	fmt.Sprintf("%x", (uint32(vendorSpecific[3])>>4)&0x0f) +
	fmt.Sprintf("%x", (uint32(vendorSpecific[3]))&0x0f)
	return tmp
	}

	// readIndications to read the indications from the OLT device
	func (om *OpenOltManager) readIndications() {
	defer logger.Errorw(nil, "Indications ended", log.Fields{})
	indications, err := om.openOltClient.EnableIndication(context.Background(), new(oop.Empty))
	if err != nil {
	logger.Errorw(nil, "Failed to read indications", log.Fields{"err": err})
	return
	}
	if indications == nil {
	logger.Errorw(nil, "Indications is nil", log.Fields{})
	return
	}

	// Create an exponential backoff around re-enabling indications. The
	// maximum elapsed time for the back off is set to 0 so that we will
	// continue to retry. The max interval defaults to 1m, but is set
	// here for code clarity
	indicationBackoff := backoff.NewExponentialBackOff()
	indicationBackoff.MaxElapsedTime = 0
	indicationBackoff.MaxInterval = 1 * time.Minute
	for {
	indication, err := indications.Recv()
	if err == io.EOF {
	logger.Infow(nil, "EOF for indications", log.Fields{"err": err})
	// Use an exponential back off to prevent getting into a tight loop
	duration := indicationBackoff.NextBackOff()
	if duration == backoff.Stop {
	// If we reach a maximum then warn and reset the backoff
	// timer and keep attempting.
	logger.Warnw(nil, "Maximum indication backoff reached, resetting backoff timer",
	log.Fields{"max_indication_backoff": indicationBackoff.MaxElapsedTime})
	indicationBackoff.Reset()
	}
	time.Sleep(indicationBackoff.NextBackOff())
	indications, err = om.openOltClient.EnableIndication(context.Background(), new(oop.Empty))
	if err != nil {
	logger.Errorw(nil, "Failed to read indications", log.Fields{"err": err})
	return
	}
	continue
	}
	if err != nil {
	logger.Infow(nil, "Failed to read from indications", log.Fields{"err": err})
	break
	}
	// Reset backoff if we have a successful receive
	indicationBackoff.Reset()
	om.handleIndication(indication)

	}
	}

	func (om OpenOltManager) handleIndication(indication oop.Indication) {
	switch indication.Data.(type) {
	case *oop.Indication_OltInd:
	logger.Info(nil, "received olt indication")
	case *oop.Indication_IntfInd:
	intfInd := indication.GetIntfInd()
	logger.Infow(nil, "Received interface indication ", log.Fields{"InterfaceInd": intfInd})
	case *oop.Indication_IntfOperInd:
	intfOperInd := indication.GetIntfOperInd()
	if intfOperInd.GetType() == "nni" {
	logger.Info(nil, "received interface oper indication for nni port")
	} else if intfOperInd.GetType() == "pon" {
	logger.Info(nil, "received interface oper indication for pon port")
	}
	/*
	case *oop.Indication_OnuDiscInd:
	onuDiscInd := indication.GetOnuDiscInd()
	logger.Infow(nil, "Received Onu discovery indication ", log.Fields{"OnuDiscInd": onuDiscInd})
	*/
	case *oop.Indication_OnuInd:
	onuInd := indication.GetOnuInd()
	logger.Infow(nil, "Received Onu indication ", log.Fields{"OnuInd": onuInd})
	case *oop.Indication_OmciInd:
	omciInd := indication.GetOmciInd()
	logger.Debugw(nil, "Received Omci indication ", log.Fields{"IntfId": omciInd.IntfId, "OnuId": omciInd.OnuId, "pkt": hex.EncodeToString(omciInd.Pkt)})
	case *oop.Indication_PktInd:
	pktInd := indication.GetPktInd()
	logger.Infow(nil, "Received packet indication ", log.Fields{"PktInd": pktInd})
	/*
	case *oop.Indication_PortStats:
	portStats := indication.GetPortStats()
	logger.Infow(nil, "Received port stats", log.Fields{"portStats": portStats})
	case *oop.Indication_FlowStats:
	flowStats := indication.GetFlowStats()
	logger.Infow(nil, "Received flow stats", log.Fields{"FlowStats": flowStats})
	*/
	case *oop.Indication_AlarmInd:
	alarmInd := indication.GetAlarmInd()
	logger.Infow(nil, "Received alarm indication ", log.Fields{"AlarmInd": alarmInd})
	}
	}

	func isPowerOfTwo(numOfOnus uint) bool {
	return (numOfOnus & (numOfOnus - 1)) == 0
	}