blob: d53891af468b68315572a9cd58c5638d3a818c24 [file] [log] [blame]
sslobodr16e41bc2019-01-18 16:22:21 -05001/*
2 * Copyright 2018-present Open Networking Foundation
3
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
7
8 * http://www.apache.org/licenses/LICENSE-2.0
9
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.
15 */
16
17package main
18
19import (
sslobodr16e41bc2019-01-18 16:22:21 -050020 "errors"
David K. Bainbridgef430cd52019-05-28 15:00:35 -070021 "flag"
Kent Hagerman334a8ce2019-05-16 16:50:33 -040022 "fmt"
Kent Hagerman92661462019-06-04 18:22:05 -040023 "google.golang.org/grpc/connectivity"
24 "google.golang.org/grpc/keepalive"
Kent Hagerman737b9e52019-06-18 16:29:33 -040025 "k8s.io/api/core/v1"
Kent Hagerman334a8ce2019-05-16 16:50:33 -040026 "math"
27 "os"
David K. Bainbridgef430cd52019-05-28 15:00:35 -070028 "path"
Kent Hagerman0ab4cb22019-04-24 13:13:35 -040029 "regexp"
sslobodr16e41bc2019-01-18 16:22:21 -050030 "strconv"
Kent Hagerman0ab4cb22019-04-24 13:13:35 -040031 "time"
sslobodr16e41bc2019-01-18 16:22:21 -050032
sslobodr16e41bc2019-01-18 16:22:21 -050033 "github.com/golang/protobuf/ptypes"
Kent Hagerman334a8ce2019-05-16 16:50:33 -040034 "github.com/golang/protobuf/ptypes/empty"
sslobodr16e41bc2019-01-18 16:22:21 -050035 "github.com/opencord/voltha-go/common/log"
David K. Bainbridgef430cd52019-05-28 15:00:35 -070036 "github.com/opencord/voltha-go/common/version"
Kent Hagerman334a8ce2019-05-16 16:50:33 -040037 "github.com/opencord/voltha-go/kafka"
William Kurkiandaa6bb22019-03-07 12:26:28 -050038 pb "github.com/opencord/voltha-protos/go/afrouter"
Kent Hagerman0ab4cb22019-04-24 13:13:35 -040039 cmn "github.com/opencord/voltha-protos/go/common"
William Kurkiandaa6bb22019-03-07 12:26:28 -050040 ic "github.com/opencord/voltha-protos/go/inter_container"
Kent Hagerman0ab4cb22019-04-24 13:13:35 -040041 vpb "github.com/opencord/voltha-protos/go/voltha"
42 "golang.org/x/net/context"
43 "google.golang.org/grpc"
44 metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
45 "k8s.io/client-go/kubernetes"
46 "k8s.io/client-go/rest"
Kent Hagermane566c2e2019-06-03 17:56:42 -040047 "k8s.io/client-go/tools/clientcmd"
sslobodr16e41bc2019-01-18 16:22:21 -050048)
49
sslobodr8e2ccb52019-02-05 09:21:47 -050050type volthaPod struct {
Kent Hagerman0ab4cb22019-04-24 13:13:35 -040051 name string
52 ipAddr string
53 node string
54 devIds map[string]struct{}
Kent Hagerman0ab4cb22019-04-24 13:13:35 -040055 backend string
sslobodr16e41bc2019-01-18 16:22:21 -050056 connection string
57}
58
David K. Bainbridgef430cd52019-05-28 15:00:35 -070059type Configuration struct {
60 DisplayVersionOnly *bool
61}
62
Kent Hagerman334a8ce2019-05-16 16:50:33 -040063var (
Kent Hagermane566c2e2019-06-03 17:56:42 -040064 // if k8s variables are undefined, will attempt to use in-cluster config
65 k8sApiServer = getStrEnv("K8S_API_SERVER", "")
66 k8sKubeConfigPath = getStrEnv("K8S_KUBE_CONFIG_PATH", "")
67
Kent Hagerman9a879af2019-07-22 17:21:09 -040068 podNamespace = getStrEnv("POD_NAMESPACE", "voltha")
69 podLabelSelector = getStrEnv("POD_LABEL_SELECTOR", "app=rw-core")
70 podAffinityGroupLabel = getStrEnv("POD_AFFINITY_GROUP_LABEL", "affinity-group")
sslobodre7ce71d2019-01-22 16:21:45 -050071
Kent Hagerman9a879af2019-07-22 17:21:09 -040072 podGrpcPort = uint64(getIntEnv("POD_GRPC_PORT", 0, math.MaxUint16, 50057))
Kent Hagerman334a8ce2019-05-16 16:50:33 -040073
Kent Hagerman92661462019-06-04 18:22:05 -040074 afrouterApiAddress = getStrEnv("AFROUTER_API_ADDRESS", "localhost:55554")
75
Kent Hagerman334a8ce2019-05-16 16:50:33 -040076 afrouterRouterName = getStrEnv("AFROUTER_ROUTER_NAME", "vcore")
Kent Hagerman9a879af2019-07-22 17:21:09 -040077 afrouterRouteName = getStrEnv("AFROUTER_ROUTE_NAME", "dev_manager")
Kent Hagerman334a8ce2019-05-16 16:50:33 -040078 afrouterRWClusterName = getStrEnv("AFROUTER_RW_CLUSTER_NAME", "vcore")
Kent Hagerman334a8ce2019-05-16 16:50:33 -040079
80 kafkaTopic = getStrEnv("KAFKA_TOPIC", "AffinityRouter")
81 kafkaClientType = getStrEnv("KAFKA_CLIENT_TYPE", "sarama")
82 kafkaHost = getStrEnv("KAFKA_HOST", "kafka")
83 kafkaPort = getIntEnv("KAFKA_PORT", 0, math.MaxUint16, 9092)
84 kafkaInstanceID = getStrEnv("KAFKA_INSTANCE_ID", "arouterd")
85)
86
87func getIntEnv(key string, min, max, defaultValue int) int {
88 if val, have := os.LookupEnv(key); have {
89 num, err := strconv.Atoi(val)
90 if err != nil || !(min <= num && num <= max) {
91 panic(fmt.Errorf("%s must be a number in the range [%d, %d]; default: %d", key, min, max, defaultValue))
92 }
93 return num
94 }
95 return defaultValue
96}
97
98func getStrEnv(key, defaultValue string) string {
99 if val, have := os.LookupEnv(key); have {
100 return val
101 }
102 return defaultValue
103}
sslobodr16e41bc2019-01-18 16:22:21 -0500104
105func newKafkaClient(clientType string, host string, port int, instanceID string) (kafka.Client, error) {
sslobodr16e41bc2019-01-18 16:22:21 -0500106 log.Infow("kafka-client-type", log.Fields{"client": clientType})
107 switch clientType {
108 case "sarama":
109 return kafka.NewSaramaClient(
110 kafka.Host(host),
111 kafka.Port(port),
112 kafka.ConsumerType(kafka.GroupCustomer),
113 kafka.ProducerReturnOnErrors(true),
114 kafka.ProducerReturnOnSuccess(true),
115 kafka.ProducerMaxRetries(6),
116 kafka.NumPartitions(3),
117 kafka.ConsumerGroupName(instanceID),
118 kafka.ConsumerGroupPrefix(instanceID),
119 kafka.AutoCreateTopic(false),
120 kafka.ProducerFlushFrequency(5),
121 kafka.ProducerRetryBackoff(time.Millisecond*30)), nil
122 }
123 return nil, errors.New("unsupported-client-type")
124}
125
sslobodr16e41bc2019-01-18 16:22:21 -0500126func k8sClientSet() *kubernetes.Clientset {
Kent Hagermane566c2e2019-06-03 17:56:42 -0400127 var config *rest.Config
128 if k8sApiServer != "" || k8sKubeConfigPath != "" {
129 // use combination of URL & local kube-config file
130 c, err := clientcmd.BuildConfigFromFlags(k8sApiServer, k8sKubeConfigPath)
131 if err != nil {
132 panic(err)
133 }
134 config = c
135 } else {
136 // use in-cluster config
137 c, err := rest.InClusterConfig()
138 if err != nil {
139 log.Errorf("Unable to load in-cluster config. Try setting K8S_API_SERVER and K8S_KUBE_CONFIG_PATH?")
140 panic(err)
141 }
142 config = c
sslobodr16e41bc2019-01-18 16:22:21 -0500143 }
144 // creates the clientset
145 clientset, err := kubernetes.NewForConfig(config)
146 if err != nil {
Kent Hagermane566c2e2019-06-03 17:56:42 -0400147 panic(err)
sslobodr16e41bc2019-01-18 16:22:21 -0500148 }
149
150 return clientset
151}
152
Kent Hagerman92661462019-06-04 18:22:05 -0400153func connect(ctx context.Context, addr string) (*grpc.ClientConn, error) {
154 log.Debugf("Trying to connect to %s", addr)
155 conn, err := grpc.DialContext(ctx, addr,
156 grpc.WithInsecure(),
157 grpc.WithBlock(),
158 grpc.WithBackoffMaxDelay(time.Second*5),
159 grpc.WithKeepaliveParams(keepalive.ClientParameters{Time: time.Second * 10, Timeout: time.Second * 5}))
160 if err == nil {
161 log.Debugf("Connection succeeded")
sslobodr16e41bc2019-01-18 16:22:21 -0500162 }
Kent Hagerman92661462019-06-04 18:22:05 -0400163 return conn, err
sslobodr16e41bc2019-01-18 16:22:21 -0500164}
165
Kent Hagerman9a879af2019-07-22 17:21:09 -0400166func getVolthaPods(cs *kubernetes.Clientset) ([]*volthaPod, error) {
167 pods, err := cs.CoreV1().Pods(podNamespace).List(metav1.ListOptions{LabelSelector: podLabelSelector})
sslobodr16e41bc2019-01-18 16:22:21 -0500168 if err != nil {
Kent Hagerman9a879af2019-07-22 17:21:09 -0400169 return nil, err
sslobodr16e41bc2019-01-18 16:22:21 -0500170 }
sslobodr16e41bc2019-01-18 16:22:21 -0500171
Kent Hagerman9a879af2019-07-22 17:21:09 -0400172 var rwPods []*volthaPod
Kent Hagerman737b9e52019-06-18 16:29:33 -0400173items:
Kent Hagerman0ab4cb22019-04-24 13:13:35 -0400174 for _, v := range pods.Items {
Kent Hagerman737b9e52019-06-18 16:29:33 -0400175 // only pods that are actually running should be considered
176 if v.Status.Phase == v1.PodRunning {
177 for _, condition := range v.Status.Conditions {
178 if condition.Status != v1.ConditionTrue {
179 continue items
180 }
181 }
182
Kent Hagerman9a879af2019-07-22 17:21:09 -0400183 if group, have := v.Labels[podAffinityGroupLabel]; have {
Kent Hagerman737b9e52019-06-18 16:29:33 -0400184 log.Debugf("Namespace: %s, PodName: %s, PodIP: %s, Host: %s\n", v.Namespace, v.Name, v.Status.PodIP, v.Spec.NodeName)
Kent Hagerman9a879af2019-07-22 17:21:09 -0400185 rwPods = append(rwPods, &volthaPod{
186 name: v.Name,
187 ipAddr: v.Status.PodIP,
188 node: v.Spec.NodeName,
189 devIds: make(map[string]struct{}),
190 backend: afrouterRWClusterName + group,
191 })
192 } else {
193 log.Warnf("Pod %s found matching % without label %", v.Name, podLabelSelector, podAffinityGroupLabel)
sslobodre7ce71d2019-01-22 16:21:45 -0500194 }
sslobodr16e41bc2019-01-18 16:22:21 -0500195 }
196 }
Kent Hagerman9a879af2019-07-22 17:21:09 -0400197 return rwPods, nil
sslobodr16e41bc2019-01-18 16:22:21 -0500198}
199
Kent Hagerman9a879af2019-07-22 17:21:09 -0400200func reconcilePodDeviceIds(ctx context.Context, pod *volthaPod, ids map[string]struct{}) {
Kent Hagerman737b9e52019-06-18 16:29:33 -0400201 ctxTimeout, _ := context.WithTimeout(ctx, time.Second*5)
202 conn, err := connect(ctxTimeout, fmt.Sprintf("%s:%d", pod.ipAddr, podGrpcPort))
sslobodr6c1689c2019-01-24 07:31:15 -0500203 if err != nil {
Kent Hagerman9a879af2019-07-22 17:21:09 -0400204 log.Debugf("Could not reconcile devices from %s, could not connect: %s", pod.name, err)
205 return
sslobodre7ce71d2019-01-22 16:21:45 -0500206 }
Kent Hagerman92661462019-06-04 18:22:05 -0400207 defer conn.Close()
Kent Hagerman334a8ce2019-05-16 16:50:33 -0400208
209 var idList cmn.IDs
210 for k := range ids {
211 idList.Items = append(idList.Items, &cmn.ID{Id: k})
212 }
213
sslobodre7ce71d2019-01-22 16:21:45 -0500214 client := vpb.NewVolthaServiceClient(conn)
Kent Hagerman737b9e52019-06-18 16:29:33 -0400215 _, err = client.ReconcileDevices(ctx, &idList)
sslobodre7ce71d2019-01-22 16:21:45 -0500216 if err != nil {
Kent Hagerman9a879af2019-07-22 17:21:09 -0400217 log.Errorf("Attempt to reconcile ids on pod %s failed: %s", pod.name, err)
218 return
sslobodre7ce71d2019-01-22 16:21:45 -0500219 }
sslobodre7ce71d2019-01-22 16:21:45 -0500220}
221
Kent Hagerman737b9e52019-06-18 16:29:33 -0400222func queryPodDeviceIds(ctx context.Context, pod *volthaPod) map[string]struct{} {
Kent Hagerman737b9e52019-06-18 16:29:33 -0400223 ctxTimeout, _ := context.WithTimeout(ctx, time.Second*5)
224 conn, err := connect(ctxTimeout, fmt.Sprintf("%s:%d", pod.ipAddr, podGrpcPort))
sslobodr6c1689c2019-01-24 07:31:15 -0500225 if err != nil {
Kent Hagerman9a879af2019-07-22 17:21:09 -0400226 log.Debugf("Could not query devices from %s, could not connect: %s", pod.name, err)
227 return nil
sslobodre7ce71d2019-01-22 16:21:45 -0500228 }
229 defer conn.Close()
Kent Hagerman9a879af2019-07-22 17:21:09 -0400230
sslobodre7ce71d2019-01-22 16:21:45 -0500231 client := vpb.NewVolthaServiceClient(conn)
Kent Hagerman737b9e52019-06-18 16:29:33 -0400232 devs, err := client.ListDeviceIds(ctx, &empty.Empty{})
sslobodre7ce71d2019-01-22 16:21:45 -0500233 if err != nil {
234 log.Error(err)
Kent Hagerman9a879af2019-07-22 17:21:09 -0400235 return nil
sslobodre7ce71d2019-01-22 16:21:45 -0500236 }
Kent Hagerman9a879af2019-07-22 17:21:09 -0400237
238 var ret = make(map[string]struct{})
Kent Hagerman0ab4cb22019-04-24 13:13:35 -0400239 for _, dv := range devs.Items {
Kent Hagerman9a879af2019-07-22 17:21:09 -0400240 ret[dv.Id] = struct{}{}
sslobodre7ce71d2019-01-22 16:21:45 -0500241 }
Kent Hagerman9a879af2019-07-22 17:21:09 -0400242 return ret
sslobodre7ce71d2019-01-22 16:21:45 -0500243}
244
Kent Hagerman9a879af2019-07-22 17:21:09 -0400245func setAffinity(ctx context.Context, client pb.ConfigurationClient, deviceId string, backend string) {
246 log.Debugf("Configuring backend %s with device id %s \n", backend, deviceId)
247 if res, err := client.SetAffinity(ctx, &pb.Affinity{
248 Router: afrouterRouterName,
249 Route: afrouterRouteName,
250 Cluster: afrouterRWClusterName,
251 Backend: backend,
252 Id: deviceId,
253 }); err != nil {
254 log.Debugf("failed affinity RPC call: %s\n", err)
sslobodr16e41bc2019-01-18 16:22:21 -0500255 } else {
Kent Hagerman9a879af2019-07-22 17:21:09 -0400256 log.Debugf("Result: %v\n", res)
sslobodr16e41bc2019-01-18 16:22:21 -0500257 }
258}
259
A R Karthickcaf4fb42019-09-05 23:16:00 +0000260func monitorDiscovery(kc kafka.Client, ctx context.Context, client pb.ConfigurationClient, ch <-chan *ic.InterContainerMessage, doneCh chan<- struct{}) {
Kent Hagerman92661462019-06-04 18:22:05 -0400261 defer close(doneCh)
A R Karthickcaf4fb42019-09-05 23:16:00 +0000262 defer kc.Stop()
Kent Hagerman92661462019-06-04 18:22:05 -0400263
Kent Hagerman9a879af2019-07-22 17:21:09 -0400264monitorLoop:
265 for {
266 select {
267 case <-ctx.Done():
A R Karthickcaf4fb42019-09-05 23:16:00 +0000268 break monitorLoop
Kent Hagerman9a879af2019-07-22 17:21:09 -0400269 case msg := <-ch:
270 log.Debug("Received a device discovery notification")
271 device := &ic.DeviceDiscovered{}
272 if err := ptypes.UnmarshalAny(msg.Body, device); err != nil {
273 log.Errorf("Could not unmarshal received notification %v", msg)
sslobodr38afd0d2019-01-21 12:31:46 -0500274 } else {
Kent Hagerman9a879af2019-07-22 17:21:09 -0400275 // somewhat hackish solution, backend is known from the first digit found in the publisher name
276 group := regexp.MustCompile(`\d`).FindString(device.Publisher)
277 if group == "" {
278 // set the affinity of the discovered device
279 setAffinity(ctx, client, device.Id, afrouterRWClusterName+group)
280 } else {
281 log.Error("backend is unknown")
282 }
sslobodr38afd0d2019-01-21 12:31:46 -0500283 }
sslobodr16e41bc2019-01-18 16:22:21 -0500284 }
sslobodr16e41bc2019-01-18 16:22:21 -0500285 }
286}
287
Kent Hagerman9a879af2019-07-22 17:21:09 -0400288func startDiscoveryMonitor(ctx context.Context, client pb.ConfigurationClient) (<-chan struct{}, error) {
Kent Hagerman92661462019-06-04 18:22:05 -0400289 doneCh := make(chan struct{})
sslobodr16e41bc2019-01-18 16:22:21 -0500290 // Connect to kafka for discovery events
Kent Hagerman334a8ce2019-05-16 16:50:33 -0400291 kc, err := newKafkaClient(kafkaClientType, kafkaHost, kafkaPort, kafkaInstanceID)
Kent Hagerman9a879af2019-07-22 17:21:09 -0400292 if err != nil {
293 panic(err)
294 }
A R Karthickcaf4fb42019-09-05 23:16:00 +0000295 if err := kc.Start(); err != nil {
296 log.Error("Could not connect to kafka, discovery disabled")
297 close(doneCh)
298 return doneCh, err
299 }
sslobodr16e41bc2019-01-18 16:22:21 -0500300
Kent Hagerman9a879af2019-07-22 17:21:09 -0400301 ch, err := kc.Subscribe(&kafka.Topic{Name: kafkaTopic})
302 if err != nil {
Kent Hagerman334a8ce2019-05-16 16:50:33 -0400303 log.Errorf("Could not subscribe to the '%s' channel, discovery disabled", kafkaTopic)
Kent Hagerman92661462019-06-04 18:22:05 -0400304 close(doneCh)
A R Karthickcaf4fb42019-09-05 23:16:00 +0000305 kc.Stop()
Kent Hagerman92661462019-06-04 18:22:05 -0400306 return doneCh, err
sslobodr16e41bc2019-01-18 16:22:21 -0500307 }
Kent Hagerman92661462019-06-04 18:22:05 -0400308
A R Karthickcaf4fb42019-09-05 23:16:00 +0000309 go monitorDiscovery(kc, ctx, client, ch, doneCh)
Kent Hagerman92661462019-06-04 18:22:05 -0400310 return doneCh, nil
sslobodr16e41bc2019-01-18 16:22:21 -0500311}
312
Kent Hagerman9a879af2019-07-22 17:21:09 -0400313// coreMonitor polls the list of devices from all RW cores, pushes these devices
314// into the affinity router, and ensures that all cores in a backend have their devices synced
315func coreMonitor(ctx context.Context, client pb.ConfigurationClient, clientset *kubernetes.Clientset) {
316 // map[backend]map[deviceId]struct{}
317 deviceOwnership := make(map[string]map[string]struct{})
Kent Hagerman92661462019-06-04 18:22:05 -0400318loop:
sslobodr16e41bc2019-01-18 16:22:21 -0500319 for {
Kent Hagerman9a879af2019-07-22 17:21:09 -0400320 // get the rw core list from k8s
321 rwPods, err := getVolthaPods(clientset)
Kent Hagerman92661462019-06-04 18:22:05 -0400322 if err != nil {
323 log.Error(err)
324 continue
325 }
326
Kent Hagerman9a879af2019-07-22 17:21:09 -0400327 // for every pod
328 for _, pod := range rwPods {
329 // get the devices for this pod's backend
330 devices, have := deviceOwnership[pod.backend]
331 if !have {
332 devices = make(map[string]struct{})
333 deviceOwnership[pod.backend] = devices
sslobodre7ce71d2019-01-22 16:21:45 -0500334 }
sslobodr8e2ccb52019-02-05 09:21:47 -0500335
Kent Hagerman9a879af2019-07-22 17:21:09 -0400336 coreDevices := queryPodDeviceIds(ctx, pod)
sslobodr16e41bc2019-01-18 16:22:21 -0500337
Kent Hagerman9a879af2019-07-22 17:21:09 -0400338 // handle devices that exist in the core, but we have just learned about
339 for deviceId := range coreDevices {
340 // if there's a new device
341 if _, have := devices[deviceId]; !have {
342 // add the device to our local list
343 devices[deviceId] = struct{}{}
344 // push the device into the affinity router
345 setAffinity(ctx, client, deviceId, pod.backend)
sslobodr8e2ccb52019-02-05 09:21:47 -0500346 }
347 }
Kent Hagerman9a879af2019-07-22 17:21:09 -0400348
349 // ensure that the core knows about all devices in its backend
350 toSync := make(map[string]struct{})
351 for deviceId := range devices {
352 // if the pod is missing any devices
353 if _, have := coreDevices[deviceId]; !have {
354 // we will reconcile them
355 toSync[deviceId] = struct{}{}
356 }
357 }
358
359 if len(toSync) != 0 {
360 reconcilePodDeviceIds(ctx, pod, toSync)
361 }
sslobodr8e2ccb52019-02-05 09:21:47 -0500362 }
Kent Hagerman9a879af2019-07-22 17:21:09 -0400363
364 select {
365 case <-ctx.Done():
366 // if we're done, exit
367 break loop
368 case <-time.After(10 * time.Second): // wait a while
369 }
sslobodre7ce71d2019-01-22 16:21:45 -0500370 }
sslobodre7ce71d2019-01-22 16:21:45 -0500371}
372
Kent Hagerman92661462019-06-04 18:22:05 -0400373// endOnClose cancels the context when the connection closes
374func connectionActiveContext(conn *grpc.ClientConn) context.Context {
375 ctx, disconnected := context.WithCancel(context.Background())
376 go func() {
377 for state := conn.GetState(); state != connectivity.TransientFailure && state != connectivity.Shutdown; state = conn.GetState() {
378 if !conn.WaitForStateChange(context.Background(), state) {
379 break
380 }
381 }
382 log.Infof("Connection to afrouter lost")
383 disconnected()
384 }()
385 return ctx
386}
sslobodr16e41bc2019-01-18 16:22:21 -0500387
Kent Hagerman92661462019-06-04 18:22:05 -0400388func main() {
David K. Bainbridgef430cd52019-05-28 15:00:35 -0700389 config := &Configuration{}
390 cmdParse := flag.NewFlagSet(path.Base(os.Args[0]), flag.ContinueOnError)
391 config.DisplayVersionOnly = cmdParse.Bool("version", false, "Print version information and exit")
392
Kent Hagerman92661462019-06-04 18:22:05 -0400393 if err := cmdParse.Parse(os.Args[1:]); err != nil {
David K. Bainbridgef430cd52019-05-28 15:00:35 -0700394 fmt.Printf("Error: %v\n", err)
395 os.Exit(1)
396 }
397
398 if *config.DisplayVersionOnly {
399 fmt.Println("VOLTHA API Server (afrouterd)")
400 fmt.Println(version.VersionInfo.String(" "))
401 return
402 }
403
sslobodr16e41bc2019-01-18 16:22:21 -0500404 // Set up logging
405 if _, err := log.SetDefaultLogger(log.JSON, 0, nil); err != nil {
406 log.With(log.Fields{"error": err}).Fatal("Cannot setup logging")
407 }
408
409 // Set up kubernetes api
410 clientset := k8sClientSet()
411
Kent Hagerman92661462019-06-04 18:22:05 -0400412 for {
413 // Connect to the affinity router
414 conn, err := connect(context.Background(), afrouterApiAddress) // This is a sidecar container so communicating over localhost
415 if err != nil {
416 panic(err)
417 }
418
419 // monitor the connection status, end context if connection is lost
420 ctx := connectionActiveContext(conn)
421
422 // set up the client
423 client := pb.NewConfigurationClient(conn)
424
Kent Hagerman9a879af2019-07-22 17:21:09 -0400425 // start the discovery monitor and core monitor
426 // these two processes do the majority of the work
427
428 log.Info("Starting discovery monitoring")
429 doneCh, _ := startDiscoveryMonitor(ctx, client)
430
431 log.Info("Starting core monitoring")
432 coreMonitor(ctx, client, clientset)
433
434 //ensure the discovery monitor to quit
435 <-doneCh
Kent Hagerman92661462019-06-04 18:22:05 -0400436
437 conn.Close()
438 }
439}