cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 1 | /* |
cbabu | 116b73f | 2019-12-10 17:56:32 +0530 | [diff] [blame] | 2 | * Copyright 2018-present Open Networking Foundation |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 3 | |
cbabu | 116b73f | 2019-12-10 17:56:32 +0530 | [diff] [blame] | 4 | * Licensed under the Apache License, Version 2.0 (the "License"); |
| 5 | * you may not use this file except in compliance with the License. |
| 6 | * You may obtain a copy of the License at |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 7 | |
cbabu | 116b73f | 2019-12-10 17:56:32 +0530 | [diff] [blame] | 8 | * http://www.apache.org/licenses/LICENSE-2.0 |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 9 | |
cbabu | 116b73f | 2019-12-10 17:56:32 +0530 | [diff] [blame] | 10 | * Unless required by applicable law or agreed to in writing, software |
| 11 | * distributed under the License is distributed on an "AS IS" BASIS, |
| 12 | * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 13 | * See the License for the specific language governing permissions and |
| 14 | * limitations under the License. |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 15 | */ |
Girish Gowdru | 6a80bbd | 2019-07-02 07:36:09 -0700 | [diff] [blame] | 16 | |
| 17 | //Package main invokes the application |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 18 | package main |
| 19 | |
| 20 | import ( |
| 21 | "context" |
| 22 | "errors" |
| 23 | "fmt" |
kdarapu | 381c690 | 2019-07-31 18:23:16 +0530 | [diff] [blame] | 24 | "os" |
| 25 | "os/signal" |
| 26 | "strconv" |
| 27 | "syscall" |
| 28 | "time" |
| 29 | |
Esin Karaman | ccb714b | 2019-11-29 15:02:06 +0000 | [diff] [blame] | 30 | "github.com/opencord/voltha-lib-go/v3/pkg/adapters/adapterif" |
kdarapu | 381c690 | 2019-07-31 18:23:16 +0530 | [diff] [blame] | 31 | |
Esin Karaman | ccb714b | 2019-11-29 15:02:06 +0000 | [diff] [blame] | 32 | "github.com/opencord/voltha-lib-go/v3/pkg/adapters" |
| 33 | com "github.com/opencord/voltha-lib-go/v3/pkg/adapters/common" |
divyadesai | a37f78b | 2020-02-07 12:41:22 +0000 | [diff] [blame] | 34 | conf "github.com/opencord/voltha-lib-go/v3/pkg/config" |
Esin Karaman | ccb714b | 2019-11-29 15:02:06 +0000 | [diff] [blame] | 35 | "github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore" |
| 36 | "github.com/opencord/voltha-lib-go/v3/pkg/kafka" |
| 37 | "github.com/opencord/voltha-lib-go/v3/pkg/log" |
| 38 | "github.com/opencord/voltha-lib-go/v3/pkg/probe" |
Scott Baker | dbd960e | 2020-02-28 08:57:51 -0800 | [diff] [blame^] | 39 | "github.com/opencord/voltha-lib-go/v3/pkg/version" |
| 40 | "github.com/opencord/voltha-openolt-adapter/internal/pkg/config" |
| 41 | ac "github.com/opencord/voltha-openolt-adapter/internal/pkg/core" |
Esin Karaman | ccb714b | 2019-11-29 15:02:06 +0000 | [diff] [blame] | 42 | ic "github.com/opencord/voltha-protos/v3/go/inter_container" |
| 43 | "github.com/opencord/voltha-protos/v3/go/voltha" |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 44 | ) |
| 45 | |
| 46 | type adapter struct { |
Girish Gowdru | 6a80bbd | 2019-07-02 07:36:09 -0700 | [diff] [blame] | 47 | instanceID string |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 48 | config *config.AdapterFlags |
| 49 | iAdapter adapters.IAdapter |
| 50 | kafkaClient kafka.Client |
| 51 | kvClient kvstore.Client |
npujar | ec5762e | 2020-01-01 14:08:48 +0530 | [diff] [blame] | 52 | kip kafka.InterContainerProxy |
kdarapu | 381c690 | 2019-07-31 18:23:16 +0530 | [diff] [blame] | 53 | coreProxy adapterif.CoreProxy |
| 54 | adapterProxy adapterif.AdapterProxy |
| 55 | eventProxy adapterif.EventProxy |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 56 | halted bool |
| 57 | exitChannel chan int |
| 58 | receiverChannels []<-chan *ic.InterContainerMessage |
| 59 | } |
| 60 | |
| 61 | func init() { |
Matteo Scandolo | 945e401 | 2019-12-12 14:16:11 -0800 | [diff] [blame] | 62 | _, _ = log.AddPackage(log.CONSOLE, log.DebugLevel, nil) |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 63 | } |
| 64 | |
| 65 | func newAdapter(cf *config.AdapterFlags) *adapter { |
| 66 | var a adapter |
Girish Gowdru | 6a80bbd | 2019-07-02 07:36:09 -0700 | [diff] [blame] | 67 | a.instanceID = cf.InstanceID |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 68 | a.config = cf |
| 69 | a.halted = false |
| 70 | a.exitChannel = make(chan int, 1) |
| 71 | a.receiverChannels = make([]<-chan *ic.InterContainerMessage, 0) |
| 72 | return &a |
| 73 | } |
| 74 | |
| 75 | func (a *adapter) start(ctx context.Context) { |
| 76 | log.Info("Starting Core Adapter components") |
| 77 | var err error |
| 78 | |
Rohan Agrawal | 828bf4e | 2019-10-22 10:13:19 +0000 | [diff] [blame] | 79 | var p *probe.Probe |
| 80 | if value := ctx.Value(probe.ProbeContextKey); value != nil { |
| 81 | if _, ok := value.(*probe.Probe); ok { |
| 82 | p = value.(*probe.Probe) |
| 83 | p.RegisterService( |
| 84 | "message-bus", |
| 85 | "kv-store", |
| 86 | "container-proxy", |
| 87 | "core-request-handler", |
| 88 | "register-with-core", |
| 89 | ) |
| 90 | } |
| 91 | } |
| 92 | |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 93 | // Setup KV Client |
| 94 | log.Debugw("create-kv-client", log.Fields{"kvstore": a.config.KVStoreType}) |
Girish Gowdru | 6a80bbd | 2019-07-02 07:36:09 -0700 | [diff] [blame] | 95 | if err = a.setKVClient(); err != nil { |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 96 | log.Fatal("error-setting-kv-client") |
| 97 | } |
| 98 | |
Rohan Agrawal | 828bf4e | 2019-10-22 10:13:19 +0000 | [diff] [blame] | 99 | if p != nil { |
| 100 | p.UpdateStatus("kv-store", probe.ServiceStatusRunning) |
| 101 | } |
| 102 | |
divyadesai | a37f78b | 2020-02-07 12:41:22 +0000 | [diff] [blame] | 103 | // Setup Log Config |
| 104 | cm := conf.NewConfigManager(a.kvClient, a.config.KVStoreType, a.config.KVStoreHost, a.config.KVStorePort, a.config.KVStoreTimeout) |
| 105 | go conf.ProcessLogConfigChange(cm, ctx) |
| 106 | |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 107 | // Setup Kafka Client |
| 108 | if a.kafkaClient, err = newKafkaClient("sarama", a.config.KafkaAdapterHost, a.config.KafkaAdapterPort); err != nil { |
| 109 | log.Fatal("Unsupported-common-client") |
| 110 | } |
| 111 | |
Rohan Agrawal | 828bf4e | 2019-10-22 10:13:19 +0000 | [diff] [blame] | 112 | if p != nil { |
| 113 | p.UpdateStatus("message-bus", probe.ServiceStatusRunning) |
| 114 | } |
| 115 | |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 116 | // Start the common InterContainer Proxy - retries indefinitely |
Rohan Agrawal | 828bf4e | 2019-10-22 10:13:19 +0000 | [diff] [blame] | 117 | if a.kip, err = a.startInterContainerProxy(ctx, -1); err != nil { |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 118 | log.Fatal("error-starting-inter-container-proxy") |
| 119 | } |
| 120 | |
| 121 | // Create the core proxy to handle requests to the Core |
| 122 | a.coreProxy = com.NewCoreProxy(a.kip, a.config.Topic, a.config.CoreTopic) |
| 123 | |
| 124 | // Create the adaptor proxy to handle request between olt and onu |
| 125 | a.adapterProxy = com.NewAdapterProxy(a.kip, "brcm_openomci_onu", a.config.CoreTopic) |
| 126 | |
Devmalya Paul | fb990a5 | 2019-07-09 10:01:49 -0400 | [diff] [blame] | 127 | // Create the event proxy to post events to KAFKA |
| 128 | a.eventProxy = com.NewEventProxy(com.MsgClient(a.kafkaClient), com.MsgTopic(kafka.Topic{Name: a.config.EventTopic})) |
| 129 | |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 130 | // Create the open OLT adapter |
kdarapu | 381c690 | 2019-07-31 18:23:16 +0530 | [diff] [blame] | 131 | if a.iAdapter, err = a.startOpenOLT(ctx, a.kip, a.coreProxy, a.adapterProxy, a.eventProxy, |
Abhilash Laxmeshwar | f9942e9 | 2020-01-07 15:32:44 +0530 | [diff] [blame] | 132 | a.config); err != nil { |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 133 | log.Fatal("error-starting-inter-container-proxy") |
| 134 | } |
| 135 | |
| 136 | // Register the core request handler |
Rohan Agrawal | 828bf4e | 2019-10-22 10:13:19 +0000 | [diff] [blame] | 137 | if err = a.setupRequestHandler(ctx, a.instanceID, a.iAdapter); err != nil { |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 138 | log.Fatal("error-setting-core-request-handler") |
| 139 | } |
| 140 | |
Girish Gowdru | 6a80bbd | 2019-07-02 07:36:09 -0700 | [diff] [blame] | 141 | // Register this adapter to the Core - retries indefinitely |
Rohan Agrawal | 828bf4e | 2019-10-22 10:13:19 +0000 | [diff] [blame] | 142 | if err = a.registerWithCore(ctx, -1); err != nil { |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 143 | log.Fatal("error-registering-with-core") |
| 144 | } |
cbabu | 95f2152 | 2019-11-13 14:25:18 +0100 | [diff] [blame] | 145 | |
cbabu | 116b73f | 2019-12-10 17:56:32 +0530 | [diff] [blame] | 146 | // check the readiness and liveliness and update the probe status |
| 147 | a.checkServicesReadiness(ctx) |
cbabu | 95f2152 | 2019-11-13 14:25:18 +0100 | [diff] [blame] | 148 | } |
| 149 | |
| 150 | /** |
| 151 | This function checks the liveliness and readiness of the kakfa and kv-client services |
| 152 | and update the status in the probe. |
| 153 | */ |
cbabu | 116b73f | 2019-12-10 17:56:32 +0530 | [diff] [blame] | 154 | func (a *adapter) checkServicesReadiness(ctx context.Context) { |
| 155 | // checks the kafka readiness |
| 156 | go a.checkKafkaReadiness(ctx) |
| 157 | |
| 158 | // checks the kv-store readiness |
| 159 | go a.checkKvStoreReadiness(ctx) |
| 160 | } |
| 161 | |
| 162 | /** |
| 163 | This function checks the liveliness and readiness of the kv-store service |
| 164 | and update the status in the probe. |
| 165 | */ |
| 166 | func (a *adapter) checkKvStoreReadiness(ctx context.Context) { |
| 167 | // dividing the live probe interval by 2 to get updated status every 30s |
| 168 | timeout := a.config.LiveProbeInterval / 2 |
| 169 | kvStoreChannel := make(chan bool, 1) |
| 170 | |
| 171 | // Default false to check the liveliness. |
| 172 | kvStoreChannel <- false |
cbabu | 95f2152 | 2019-11-13 14:25:18 +0100 | [diff] [blame] | 173 | for { |
cbabu | 116b73f | 2019-12-10 17:56:32 +0530 | [diff] [blame] | 174 | timeoutTimer := time.NewTimer(timeout) |
| 175 | select { |
| 176 | case liveliness := <-kvStoreChannel: |
| 177 | if !liveliness { |
| 178 | // kv-store not reachable or down, updating the status to not ready state |
| 179 | probe.UpdateStatusFromContext(ctx, "kv-store", probe.ServiceStatusNotReady) |
| 180 | timeout = a.config.NotLiveProbeInterval |
| 181 | } else { |
| 182 | // kv-store is reachable , updating the status to running state |
| 183 | probe.UpdateStatusFromContext(ctx, "kv-store", probe.ServiceStatusRunning) |
| 184 | timeout = a.config.LiveProbeInterval / 2 |
| 185 | } |
| 186 | // Check if the timer has expired or not |
| 187 | if !timeoutTimer.Stop() { |
| 188 | <-timeoutTimer.C |
| 189 | } |
| 190 | case <-timeoutTimer.C: |
| 191 | // Check the status of the kv-store |
| 192 | log.Info("kv-store liveliness-recheck") |
npujar | ec5762e | 2020-01-01 14:08:48 +0530 | [diff] [blame] | 193 | if a.kvClient.IsConnectionUp(ctx) { |
cbabu | 116b73f | 2019-12-10 17:56:32 +0530 | [diff] [blame] | 194 | kvStoreChannel <- true |
| 195 | } else { |
| 196 | kvStoreChannel <- false |
| 197 | } |
cbabu | 95f2152 | 2019-11-13 14:25:18 +0100 | [diff] [blame] | 198 | } |
cbabu | 116b73f | 2019-12-10 17:56:32 +0530 | [diff] [blame] | 199 | } |
| 200 | } |
| 201 | |
| 202 | /** |
| 203 | This function checks the liveliness and readiness of the kafka service |
| 204 | and update the status in the probe. |
| 205 | */ |
| 206 | func (a *adapter) checkKafkaReadiness(ctx context.Context) { |
| 207 | livelinessChannel := a.kafkaClient.EnableLivenessChannel(true) |
Scott Baker | 86fce9a | 2019-12-12 09:47:17 -0800 | [diff] [blame] | 208 | healthinessChannel := a.kafkaClient.EnableHealthinessChannel(true) |
cbabu | 116b73f | 2019-12-10 17:56:32 +0530 | [diff] [blame] | 209 | timeout := a.config.LiveProbeInterval |
Scott Baker | e701b86 | 2020-02-20 16:19:16 -0800 | [diff] [blame] | 210 | failed := false |
cbabu | 116b73f | 2019-12-10 17:56:32 +0530 | [diff] [blame] | 211 | for { |
| 212 | timeoutTimer := time.NewTimer(timeout) |
| 213 | |
| 214 | select { |
Scott Baker | 86fce9a | 2019-12-12 09:47:17 -0800 | [diff] [blame] | 215 | case healthiness := <-healthinessChannel: |
| 216 | if !healthiness { |
Scott Baker | e701b86 | 2020-02-20 16:19:16 -0800 | [diff] [blame] | 217 | // This will eventually cause K8s to restart the container, and will do |
| 218 | // so in a way that allows cleanup to continue, rather than an immediate |
| 219 | // panic and exit here. |
| 220 | probe.UpdateStatusFromContext(ctx, "message-bus", probe.ServiceStatusFailed) |
| 221 | failed = true |
| 222 | } |
| 223 | // Check if the timer has expired or not |
| 224 | if !timeoutTimer.Stop() { |
| 225 | <-timeoutTimer.C |
Scott Baker | 86fce9a | 2019-12-12 09:47:17 -0800 | [diff] [blame] | 226 | } |
cbabu | 116b73f | 2019-12-10 17:56:32 +0530 | [diff] [blame] | 227 | case liveliness := <-livelinessChannel: |
Scott Baker | e701b86 | 2020-02-20 16:19:16 -0800 | [diff] [blame] | 228 | if failed { |
| 229 | // Failures of the message bus are permanent and can't ever be recovered from, |
| 230 | // so make sure we never inadvertently reset a failed state back to unready. |
| 231 | } else if !liveliness { |
cbabu | 116b73f | 2019-12-10 17:56:32 +0530 | [diff] [blame] | 232 | // kafka not reachable or down, updating the status to not ready state |
| 233 | probe.UpdateStatusFromContext(ctx, "message-bus", probe.ServiceStatusNotReady) |
| 234 | timeout = a.config.NotLiveProbeInterval |
| 235 | } else { |
| 236 | // kafka is reachable , updating the status to running state |
| 237 | probe.UpdateStatusFromContext(ctx, "message-bus", probe.ServiceStatusRunning) |
| 238 | timeout = a.config.LiveProbeInterval |
| 239 | } |
| 240 | // Check if the timer has expired or not |
| 241 | if !timeoutTimer.Stop() { |
| 242 | <-timeoutTimer.C |
| 243 | } |
| 244 | case <-timeoutTimer.C: |
| 245 | log.Info("kafka-proxy-liveness-recheck") |
| 246 | // send the liveness probe in a goroutine; we don't want to deadlock ourselves as |
| 247 | // the liveness probe may wait (and block) writing to our channel. |
| 248 | err := a.kafkaClient.SendLiveness() |
| 249 | if err != nil { |
| 250 | // Catch possible error case if sending liveness after Sarama has been stopped. |
| 251 | log.Warnw("error-kafka-send-liveness", log.Fields{"error": err}) |
| 252 | } |
cbabu | 95f2152 | 2019-11-13 14:25:18 +0100 | [diff] [blame] | 253 | } |
cbabu | 95f2152 | 2019-11-13 14:25:18 +0100 | [diff] [blame] | 254 | } |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 255 | } |
| 256 | |
npujar | ec5762e | 2020-01-01 14:08:48 +0530 | [diff] [blame] | 257 | func (a *adapter) stop(ctx context.Context) { |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 258 | // Stop leadership tracking |
Girish Gowdru | 6a80bbd | 2019-07-02 07:36:09 -0700 | [diff] [blame] | 259 | a.halted = true |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 260 | |
| 261 | // send exit signal |
Girish Gowdru | 6a80bbd | 2019-07-02 07:36:09 -0700 | [diff] [blame] | 262 | a.exitChannel <- 0 |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 263 | |
| 264 | // Cleanup - applies only if we had a kvClient |
Girish Gowdru | 6a80bbd | 2019-07-02 07:36:09 -0700 | [diff] [blame] | 265 | if a.kvClient != nil { |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 266 | // Release all reservations |
npujar | ec5762e | 2020-01-01 14:08:48 +0530 | [diff] [blame] | 267 | if err := a.kvClient.ReleaseAllReservations(ctx); err != nil { |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 268 | log.Infow("fail-to-release-all-reservations", log.Fields{"error": err}) |
| 269 | } |
| 270 | // Close the DB connection |
Girish Gowdru | 6a80bbd | 2019-07-02 07:36:09 -0700 | [diff] [blame] | 271 | a.kvClient.Close() |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 272 | } |
| 273 | |
Scott Baker | e701b86 | 2020-02-20 16:19:16 -0800 | [diff] [blame] | 274 | if a.kip != nil { |
| 275 | a.kip.Stop() |
| 276 | } |
| 277 | |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 278 | // TODO: More cleanup |
| 279 | } |
| 280 | |
Girish Gowdru | 6a80bbd | 2019-07-02 07:36:09 -0700 | [diff] [blame] | 281 | func newKVClient(storeType, address string, timeout int) (kvstore.Client, error) { |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 282 | |
| 283 | log.Infow("kv-store-type", log.Fields{"store": storeType}) |
| 284 | switch storeType { |
| 285 | case "consul": |
| 286 | return kvstore.NewConsulClient(address, timeout) |
| 287 | case "etcd": |
| 288 | return kvstore.NewEtcdClient(address, timeout) |
| 289 | } |
| 290 | return nil, errors.New("unsupported-kv-store") |
| 291 | } |
| 292 | |
Girish Gowdru | 6a80bbd | 2019-07-02 07:36:09 -0700 | [diff] [blame] | 293 | func newKafkaClient(clientType, host string, port int) (kafka.Client, error) { |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 294 | |
| 295 | log.Infow("common-client-type", log.Fields{"client": clientType}) |
| 296 | switch clientType { |
| 297 | case "sarama": |
| 298 | return kafka.NewSaramaClient( |
| 299 | kafka.Host(host), |
| 300 | kafka.Port(port), |
| 301 | kafka.ProducerReturnOnErrors(true), |
| 302 | kafka.ProducerReturnOnSuccess(true), |
| 303 | kafka.ProducerMaxRetries(6), |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame] | 304 | kafka.ProducerRetryBackoff(time.Millisecond*30), |
| 305 | kafka.MetadatMaxRetries(15)), nil |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 306 | } |
Girish Gowdru | 6a80bbd | 2019-07-02 07:36:09 -0700 | [diff] [blame] | 307 | |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 308 | return nil, errors.New("unsupported-client-type") |
| 309 | } |
| 310 | |
| 311 | func (a *adapter) setKVClient() error { |
| 312 | addr := a.config.KVStoreHost + ":" + strconv.Itoa(a.config.KVStorePort) |
| 313 | client, err := newKVClient(a.config.KVStoreType, addr, a.config.KVStoreTimeout) |
| 314 | if err != nil { |
| 315 | a.kvClient = nil |
| 316 | log.Error(err) |
| 317 | return err |
| 318 | } |
| 319 | a.kvClient = client |
divyadesai | a37f78b | 2020-02-07 12:41:22 +0000 | [diff] [blame] | 320 | |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 321 | return nil |
| 322 | } |
| 323 | |
npujar | ec5762e | 2020-01-01 14:08:48 +0530 | [diff] [blame] | 324 | func (a *adapter) startInterContainerProxy(ctx context.Context, retries int) (kafka.InterContainerProxy, error) { |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 325 | log.Infow("starting-intercontainer-messaging-proxy", log.Fields{"host": a.config.KafkaAdapterHost, |
| 326 | "port": a.config.KafkaAdapterPort, "topic": a.config.Topic}) |
| 327 | var err error |
npujar | ec5762e | 2020-01-01 14:08:48 +0530 | [diff] [blame] | 328 | kip := kafka.NewInterContainerProxy( |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 329 | kafka.InterContainerHost(a.config.KafkaAdapterHost), |
| 330 | kafka.InterContainerPort(a.config.KafkaAdapterPort), |
| 331 | kafka.MsgClient(a.kafkaClient), |
npujar | ec5762e | 2020-01-01 14:08:48 +0530 | [diff] [blame] | 332 | kafka.DefaultTopic(&kafka.Topic{Name: a.config.Topic})) |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 333 | count := 0 |
| 334 | for { |
| 335 | if err = kip.Start(); err != nil { |
| 336 | log.Warnw("error-starting-messaging-proxy", log.Fields{"error": err}) |
| 337 | if retries == count { |
| 338 | return nil, err |
| 339 | } |
| 340 | count = +1 |
Girish Gowdru | 6a80bbd | 2019-07-02 07:36:09 -0700 | [diff] [blame] | 341 | // Take a nap before retrying |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 342 | time.Sleep(2 * time.Second) |
| 343 | } else { |
| 344 | break |
| 345 | } |
| 346 | } |
Rohan Agrawal | 828bf4e | 2019-10-22 10:13:19 +0000 | [diff] [blame] | 347 | probe.UpdateStatusFromContext(ctx, "container-proxy", probe.ServiceStatusRunning) |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 348 | log.Info("common-messaging-proxy-created") |
| 349 | return kip, nil |
| 350 | } |
| 351 | |
npujar | ec5762e | 2020-01-01 14:08:48 +0530 | [diff] [blame] | 352 | func (a *adapter) startOpenOLT(ctx context.Context, kip kafka.InterContainerProxy, |
Abhilash Laxmeshwar | f9942e9 | 2020-01-07 15:32:44 +0530 | [diff] [blame] | 353 | cp adapterif.CoreProxy, ap adapterif.AdapterProxy, ep adapterif.EventProxy, |
| 354 | cfg *config.AdapterFlags) (*ac.OpenOLT, error) { |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 355 | log.Info("starting-open-olt") |
| 356 | var err error |
Abhilash Laxmeshwar | f9942e9 | 2020-01-07 15:32:44 +0530 | [diff] [blame] | 357 | sOLT := ac.NewOpenOLT(ctx, a.kip, cp, ap, ep, cfg) |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 358 | |
| 359 | if err = sOLT.Start(ctx); err != nil { |
| 360 | log.Fatalw("error-starting-messaging-proxy", log.Fields{"error": err}) |
| 361 | return nil, err |
| 362 | } |
| 363 | |
| 364 | log.Info("open-olt-started") |
| 365 | return sOLT, nil |
| 366 | } |
| 367 | |
Rohan Agrawal | 828bf4e | 2019-10-22 10:13:19 +0000 | [diff] [blame] | 368 | func (a *adapter) setupRequestHandler(ctx context.Context, coreInstanceID string, iadapter adapters.IAdapter) error { |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 369 | log.Info("setting-request-handler") |
Girish Gowdru | 6a80bbd | 2019-07-02 07:36:09 -0700 | [diff] [blame] | 370 | requestProxy := com.NewRequestHandlerProxy(coreInstanceID, iadapter, a.coreProxy) |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 371 | if err := a.kip.SubscribeWithRequestHandlerInterface(kafka.Topic{Name: a.config.Topic}, requestProxy); err != nil { |
| 372 | log.Errorw("request-handler-setup-failed", log.Fields{"error": err}) |
| 373 | return err |
| 374 | |
| 375 | } |
Rohan Agrawal | 828bf4e | 2019-10-22 10:13:19 +0000 | [diff] [blame] | 376 | probe.UpdateStatusFromContext(ctx, "core-request-handler", probe.ServiceStatusRunning) |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 377 | log.Info("request-handler-setup-done") |
| 378 | return nil |
| 379 | } |
| 380 | |
Rohan Agrawal | 828bf4e | 2019-10-22 10:13:19 +0000 | [diff] [blame] | 381 | func (a *adapter) registerWithCore(ctx context.Context, retries int) error { |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 382 | log.Info("registering-with-core") |
Girish Gowdru | 0c588b2 | 2019-04-23 23:24:56 -0400 | [diff] [blame] | 383 | adapterDescription := &voltha.Adapter{Id: "openolt", // Unique name for the device type |
Matt Jeanneret | f880eb6 | 2019-07-16 20:08:03 -0400 | [diff] [blame] | 384 | Vendor: "VOLTHA OpenOLT", |
| 385 | Version: version.VersionInfo.Version} |
Girish Gowdru | 0c588b2 | 2019-04-23 23:24:56 -0400 | [diff] [blame] | 386 | types := []*voltha.DeviceType{{Id: "openolt", |
Girish Gowdru | 6a80bbd | 2019-07-02 07:36:09 -0700 | [diff] [blame] | 387 | Adapter: "openolt", // Name of the adapter that handles device type |
Girish Gowdru | 0c588b2 | 2019-04-23 23:24:56 -0400 | [diff] [blame] | 388 | AcceptsBulkFlowUpdate: false, // Currently openolt adapter does not support bulk flow handling |
| 389 | AcceptsAddRemoveFlowUpdates: true}} |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 390 | deviceTypes := &voltha.DeviceTypes{Items: types} |
| 391 | count := 0 |
| 392 | for { |
Girish Gowdru | 6a80bbd | 2019-07-02 07:36:09 -0700 | [diff] [blame] | 393 | if err := a.coreProxy.RegisterAdapter(context.TODO(), adapterDescription, deviceTypes); err != nil { |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 394 | log.Warnw("registering-with-core-failed", log.Fields{"error": err}) |
| 395 | if retries == count { |
| 396 | return err |
| 397 | } |
Girish Gowdru | 6a80bbd | 2019-07-02 07:36:09 -0700 | [diff] [blame] | 398 | count++ |
| 399 | // Take a nap before retrying |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 400 | time.Sleep(2 * time.Second) |
| 401 | } else { |
| 402 | break |
| 403 | } |
| 404 | } |
Rohan Agrawal | 828bf4e | 2019-10-22 10:13:19 +0000 | [diff] [blame] | 405 | probe.UpdateStatusFromContext(ctx, "register-with-core", probe.ServiceStatusRunning) |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 406 | log.Info("registered-with-core") |
| 407 | return nil |
| 408 | } |
| 409 | |
| 410 | func waitForExit() int { |
| 411 | signalChannel := make(chan os.Signal, 1) |
| 412 | signal.Notify(signalChannel, |
| 413 | syscall.SIGHUP, |
| 414 | syscall.SIGINT, |
| 415 | syscall.SIGTERM, |
| 416 | syscall.SIGQUIT) |
| 417 | |
| 418 | exitChannel := make(chan int) |
| 419 | |
| 420 | go func() { |
| 421 | s := <-signalChannel |
| 422 | switch s { |
| 423 | case syscall.SIGHUP, |
| 424 | syscall.SIGINT, |
| 425 | syscall.SIGTERM, |
| 426 | syscall.SIGQUIT: |
| 427 | log.Infow("closing-signal-received", log.Fields{"signal": s}) |
| 428 | exitChannel <- 0 |
| 429 | default: |
| 430 | log.Infow("unexpected-signal-received", log.Fields{"signal": s}) |
| 431 | exitChannel <- 1 |
| 432 | } |
| 433 | }() |
| 434 | |
| 435 | code := <-exitChannel |
| 436 | return code |
| 437 | } |
| 438 | |
| 439 | func printBanner() { |
David K. Bainbridge | 794735f | 2020-02-11 21:01:37 -0800 | [diff] [blame] | 440 | fmt.Println(` ____ ____ _ _______ `) |
| 441 | fmt.Println(` / _ \ / __ \| | |__ __|`) |
| 442 | fmt.Println(` | | | |_ __ ___ _ __ | | | | | | | `) |
| 443 | fmt.Println(` | | | | '_ \ / _ \ '_ \ | | | | | | | `) |
| 444 | fmt.Println(` | |__| | |_) | __/ | | || |__| | |____| | `) |
| 445 | fmt.Println(` \____/| .__/ \___|_| |_| \____/|______|_| `) |
| 446 | fmt.Println(` | | `) |
| 447 | fmt.Println(` |_| `) |
| 448 | fmt.Println(` `) |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 449 | } |
| 450 | |
Matt Jeanneret | f880eb6 | 2019-07-16 20:08:03 -0400 | [diff] [blame] | 451 | func printVersion() { |
| 452 | fmt.Println("VOLTHA OpenOLT Adapter") |
| 453 | fmt.Println(version.VersionInfo.String(" ")) |
| 454 | } |
| 455 | |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 456 | func main() { |
| 457 | start := time.Now() |
| 458 | |
| 459 | cf := config.NewAdapterFlags() |
| 460 | cf.ParseCommandArguments() |
| 461 | |
Girish Gowdru | 6a80bbd | 2019-07-02 07:36:09 -0700 | [diff] [blame] | 462 | // Setup logging |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 463 | |
Rohan Agrawal | 02f784d | 2020-02-14 09:34:02 +0000 | [diff] [blame] | 464 | logLevel, err := log.StringToLogLevel(cf.LogLevel) |
| 465 | if err != nil { |
| 466 | log.Fatalf("Cannot setup logging, %s", err) |
| 467 | } |
Rohan Agrawal | 2488f19 | 2020-01-31 09:26:55 +0000 | [diff] [blame] | 468 | |
Girish Gowdru | 6a80bbd | 2019-07-02 07:36:09 -0700 | [diff] [blame] | 469 | // Setup default logger - applies for packages that do not have specific logger set |
Rohan Agrawal | 02f784d | 2020-02-14 09:34:02 +0000 | [diff] [blame] | 470 | if _, err := log.SetDefaultLogger(log.JSON, logLevel, log.Fields{"instanceId": cf.InstanceID}); err != nil { |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 471 | log.With(log.Fields{"error": err}).Fatal("Cannot setup logging") |
| 472 | } |
| 473 | |
| 474 | // Update all loggers (provisionned via init) with a common field |
Hardik Windlass | b9c869b | 2019-10-10 08:34:32 +0000 | [diff] [blame] | 475 | if err := log.UpdateAllLoggers(log.Fields{"instanceId": cf.InstanceID}); err != nil { |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 476 | log.With(log.Fields{"error": err}).Fatal("Cannot setup logging") |
| 477 | } |
| 478 | |
Rohan Agrawal | 02f784d | 2020-02-14 09:34:02 +0000 | [diff] [blame] | 479 | log.SetAllLogLevel(logLevel) |
Rohan Agrawal | 93bced3 | 2020-02-11 10:16:01 +0000 | [diff] [blame] | 480 | |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 481 | defer log.CleanUp() |
| 482 | |
Matt Jeanneret | f880eb6 | 2019-07-16 20:08:03 -0400 | [diff] [blame] | 483 | // Print version / build information and exit |
| 484 | if cf.DisplayVersionOnly { |
| 485 | printVersion() |
| 486 | return |
| 487 | } |
| 488 | |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 489 | // Print banner if specified |
| 490 | if cf.Banner { |
| 491 | printBanner() |
| 492 | } |
| 493 | |
| 494 | log.Infow("config", log.Fields{"config": *cf}) |
| 495 | |
| 496 | ctx, cancel := context.WithCancel(context.Background()) |
| 497 | defer cancel() |
| 498 | |
| 499 | ad := newAdapter(cf) |
Rohan Agrawal | 828bf4e | 2019-10-22 10:13:19 +0000 | [diff] [blame] | 500 | |
| 501 | p := &probe.Probe{} |
| 502 | go p.ListenAndServe(fmt.Sprintf("%s:%d", ad.config.ProbeHost, ad.config.ProbePort)) |
| 503 | |
| 504 | probeCtx := context.WithValue(ctx, probe.ProbeContextKey, p) |
| 505 | |
| 506 | go ad.start(probeCtx) |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 507 | |
| 508 | code := waitForExit() |
| 509 | log.Infow("received-a-closing-signal", log.Fields{"code": code}) |
| 510 | |
| 511 | // Cleanup before leaving |
npujar | ec5762e | 2020-01-01 14:08:48 +0530 | [diff] [blame] | 512 | ad.stop(ctx) |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 513 | |
| 514 | elapsed := time.Since(start) |
Hardik Windlass | b9c869b | 2019-10-10 08:34:32 +0000 | [diff] [blame] | 515 | log.Infow("run-time", log.Fields{"instanceId": ad.config.InstanceID, "time": elapsed / time.Second}) |
cuilin2018 | 7b2a8c3 | 2019-03-26 19:52:28 -0700 | [diff] [blame] | 516 | } |