[VOL-1825] Implemetation of the reconcile feature in the Core
This commit implements the reconcile feature in the Core as
well as the necessary changes in the GO libraries. Changes were
also made in the simulated OLT and ONU adapters to react to a
reconcile request following an adapter restart. These changes
were tested in a 1-node deployment (1 core pair) and 3-nodes
deployment (3 core-pairs but single adapter instances). Test
was also done using the openolt to ensure the reconcile request
does not cause panic in the Core or Openolt since the reconcile
feature is not implemented in Openolt.
Change-Id: I178d619fbcfcc6caa0ee3abfd70fc03c384c5fea
diff --git a/rw_core/core/adapter_request_handler.go b/rw_core/core/adapter_request_handler.go
index 1a00db8..33d14a9 100644
--- a/rw_core/core/adapter_request_handler.go
+++ b/rw_core/core/adapter_request_handler.go
@@ -148,7 +148,7 @@
if txn, err := rhp.acquireRequest(transactionID.Val); err != nil {
log.Debugw("Another core handled the request", log.Fields{"transactionID": transactionID})
// Update our adapters in memory
- go rhp.adapterMgr.updateAdaptersAndDevicetypesInMemory()
+ go rhp.adapterMgr.updateAdaptersAndDevicetypesInMemory(adapter)
// returning nil, nil instructs the callee to ignore this request
return nil, nil
} else {
@@ -1135,3 +1135,48 @@
//}
return new(empty.Empty), nil
}
+
+func (rhp *AdapterRequestHandlerProxy) ReconcileChildDevices(args []*ic.Argument) (*empty.Empty, error) {
+ if len(args) < 2 {
+ log.Warn("invalid-number-of-args", log.Fields{"args": args})
+ err := errors.New("invalid-number-of-args")
+ return nil, err
+ }
+ parentDeviceId := &voltha.ID{}
+ transactionID := &ic.StrType{}
+ for _, arg := range args {
+ switch arg.Key {
+ case "parent_device_id":
+ if err := ptypes.UnmarshalAny(arg.Value, parentDeviceId); err != nil {
+ log.Warnw("cannot-unmarshal-device-id", log.Fields{"error": err})
+ return nil, err
+ }
+ case kafka.TransactionKey:
+ if err := ptypes.UnmarshalAny(arg.Value, transactionID); err != nil {
+ log.Warnw("cannot-unmarshal-transaction-ID", log.Fields{"error": err})
+ return nil, err
+ }
+ }
+ }
+ log.Debugw("ReconcileChildDevices", log.Fields{"deviceId": parentDeviceId.Id, "transactionID": transactionID.Val})
+
+ // Try to grab the transaction as this core may be competing with another Core
+ if rhp.competeForTransaction() {
+ if txn, err := rhp.takeRequestOwnership(transactionID.Val, parentDeviceId.Id); err != nil {
+ log.Debugw("Another core handled the request", log.Fields{"transactionID": transactionID})
+ // returning nil, nil instructs the callee to ignore this request
+ return nil, nil
+ } else {
+ defer txn.Close()
+ }
+ }
+
+ if rhp.TestMode { // Execute only for test cases
+ return nil, nil
+ }
+
+ // Run it in its own routine
+ go rhp.deviceMgr.reconcileChildDevices(parentDeviceId.Id)
+
+ return new(empty.Empty), nil
+}