blob: 53db2642a42a46eceb486a26fab4fa2521410624 [file] [log] [blame]
Kent Hagerman2f0d0552020-04-23 17:28:52 -04001/*
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 core
18
19import (
20 "context"
21 "errors"
22 "time"
23
24 "github.com/opencord/voltha-lib-go/v3/pkg/db"
25 "github.com/opencord/voltha-lib-go/v3/pkg/db/kvstore"
26 "github.com/opencord/voltha-lib-go/v3/pkg/log"
27 "github.com/opencord/voltha-lib-go/v3/pkg/probe"
28 "google.golang.org/grpc/codes"
29 "google.golang.org/grpc/status"
30)
31
Neha Sharma7d6f3a92020-04-14 15:26:22 +000032func newKVClient(storeType string, address string, timeout time.Duration) (kvstore.Client, error) {
Kent Hagerman2f0d0552020-04-23 17:28:52 -040033 logger.Infow("kv-store-type", log.Fields{"store": storeType})
34 switch storeType {
35 case "consul":
36 return kvstore.NewConsulClient(address, timeout)
37 case "etcd":
38 return kvstore.NewEtcdClient(address, timeout, log.FatalLevel)
39 }
40 return nil, errors.New("unsupported-kv-store")
41}
42
43func stopKVClient(ctx context.Context, kvClient kvstore.Client) {
44 // Release all reservations
45 if err := kvClient.ReleaseAllReservations(ctx); err != nil {
46 logger.Infow("fail-to-release-all-reservations", log.Fields{"error": err})
47 }
48 // Close the DB connection
49 kvClient.Close()
50}
51
52// waitUntilKVStoreReachableOrMaxTries will wait until it can connect to a KV store or until maxtries has been reached
53func waitUntilKVStoreReachableOrMaxTries(ctx context.Context, kvClient kvstore.Client, maxRetries int, retryInterval time.Duration) error {
54 logger.Infow("verifying-KV-store-connectivity", log.Fields{"retries": maxRetries, "retryInterval": retryInterval})
55 count := 0
56 for {
57 if !kvClient.IsConnectionUp(ctx) {
58 logger.Info("KV-store-unreachable")
59 if maxRetries != -1 {
60 if count >= maxRetries {
61 return status.Error(codes.Unavailable, "kv store unreachable")
62 }
63 }
64 count++
65
66 // Take a nap before retrying
67 select {
68 case <-ctx.Done():
69 //ctx canceled
70 return ctx.Err()
71 case <-time.After(retryInterval):
72 }
73 logger.Infow("retry-KV-store-connectivity", log.Fields{"retryCount": count, "maxRetries": maxRetries, "retryInterval": retryInterval})
74 } else {
75 break
76 }
77 }
78 probe.UpdateStatusFromContext(ctx, "kv-store", probe.ServiceStatusRunning)
79 logger.Info("KV-store-reachable")
80 return nil
81}
82
83/*
84 * Thread to monitor kvstore Liveness (connection status)
85 *
86 * This function constantly monitors Liveness State of kvstore as reported
87 * periodically by backend and updates the Status of kv-store service registered
88 * with rw_core probe.
89 *
90 * If no liveness event has been seen within a timeout, then the thread will
91 * perform a "liveness" check attempt, which will in turn trigger a liveness event on
92 * the liveness channel, true or false depending on whether the attempt succeeded.
93 *
94 * The gRPC server in turn monitors the state of the readiness probe and will
95 * start issuing UNAVAILABLE response while the probe is not ready.
96 */
97func monitorKVStoreLiveness(ctx context.Context, backend *db.Backend, liveProbeInterval, notLiveProbeInterval time.Duration) {
98 logger.Info("start-monitoring-kvstore-liveness")
99
100 // Instruct backend to create Liveness channel for transporting state updates
101 livenessChannel := backend.EnableLivenessChannel()
102
103 logger.Debug("enabled-kvstore-liveness-channel")
104
105 // Default state for kvstore is alive for rw_core
106 timeout := liveProbeInterval
107loop:
108 for {
109 timeoutTimer := time.NewTimer(timeout)
110 select {
111
112 case liveness := <-livenessChannel:
113 logger.Debugw("received-liveness-change-notification", log.Fields{"liveness": liveness})
114
115 if !liveness {
116 probe.UpdateStatusFromContext(ctx, "kv-store", probe.ServiceStatusNotReady)
117 logger.Info("kvstore-set-server-notready")
118
119 timeout = notLiveProbeInterval
120
121 } else {
122 probe.UpdateStatusFromContext(ctx, "kv-store", probe.ServiceStatusRunning)
123 logger.Info("kvstore-set-server-ready")
124
125 timeout = liveProbeInterval
126 }
127
128 if !timeoutTimer.Stop() {
129 <-timeoutTimer.C
130 }
131
132 case <-ctx.Done():
133 break loop
134
135 case <-timeoutTimer.C:
136 logger.Info("kvstore-perform-liveness-check-on-timeout")
137
138 // Trigger Liveness check if no liveness update received within the timeout period.
139 // The Liveness check will push Live state to same channel which this routine is
140 // reading and processing. This, do it asynchronously to avoid blocking for
141 // backend response and avoid any possibility of deadlock
142 go backend.PerformLivenessCheck(ctx)
143 }
144 }
145}