VOL-1739 [Create Kubernetes Probes for Simulated OLT Adapter]
Change-Id: Ib0ef80daf7843b27f95d87894a829a4826190881
diff --git a/cmd/simulated_olt/main.go b/cmd/simulated_olt/main.go
index f324939..e67c6ca 100644
--- a/cmd/simulated_olt/main.go
+++ b/cmd/simulated_olt/main.go
@@ -22,6 +22,7 @@
"github.com/opencord/voltha-go/adapters"
com "github.com/opencord/voltha-go/adapters/common"
"github.com/opencord/voltha-go/common/log"
+ "github.com/opencord/voltha-go/common/probe"
"github.com/opencord/voltha-go/common/version"
"github.com/opencord/voltha-go/db/kvstore"
"github.com/opencord/voltha-go/kafka"
@@ -67,19 +68,41 @@
log.Info("Starting Core Adapter components")
var err error
+ var p *probe.Probe
+ if value := ctx.Value(probe.ProbeContextKey); value != nil {
+ if _, ok := value.(*probe.Probe); ok {
+ p = value.(*probe.Probe)
+ p.RegisterService(
+ "message-bus",
+ "kv-store",
+ "container-proxy",
+ "core-request-handler",
+ "register-with-core",
+ )
+ }
+ }
+
// Setup KV Client
log.Debugw("create-kv-client", log.Fields{"kvstore": a.config.KVStoreType})
if err := a.setKVClient(); err != nil {
log.Fatal("error-setting-kv-client")
}
+ if p != nil {
+ p.UpdateStatus("kv-store", probe.ServiceStatusRunning)
+ }
+
// Setup Kafka Client
if a.kafkaClient, err = newKafkaClient("sarama", a.config.KafkaAdapterHost, a.config.KafkaAdapterPort); err != nil {
log.Fatal("Unsupported-common-client")
}
+ if p != nil {
+ p.UpdateStatus("message-bus", probe.ServiceStatusRunning)
+ }
+
// Start the common InterContainer Proxy - retries indefinitely
- if a.kip, err = a.startInterContainerProxy(-1); err != nil {
+ if a.kip, err = a.startInterContainerProxy(ctx, -1); err != nil {
log.Fatal("error-starting-inter-container-proxy")
}
@@ -92,12 +115,12 @@
}
// Register the core request handler
- if err = a.setupRequestHandler(a.instanceId, a.iAdapter, a.coreProxy); err != nil {
+ if err = a.setupRequestHandler(ctx, a.instanceId, a.iAdapter, a.coreProxy); err != nil {
log.Fatal("error-setting-core-request-handler")
}
// Register this adapter to the Core - retries indefinitely
- if err = a.registerWithCore(-1); err != nil {
+ if err = a.registerWithCore(ctx, -1); err != nil {
log.Fatal("error-registering-with-core")
}
}
@@ -173,7 +196,7 @@
}
}
-func (a *adapter) startInterContainerProxy(retries int) (*kafka.InterContainerProxy, error) {
+func (a *adapter) startInterContainerProxy(ctx context.Context, retries int) (*kafka.InterContainerProxy, error) {
log.Infow("starting-intercontainer-messaging-proxy", log.Fields{"host": a.config.KafkaAdapterHost,
"port": a.config.KafkaAdapterPort, "topic": a.config.Topic})
var err error
@@ -201,6 +224,7 @@
}
}
+ probe.UpdateStatusFromContext(ctx, "container-proxy", probe.ServiceStatusRunning)
log.Info("common-messaging-proxy-created")
return kip, nil
}
@@ -219,7 +243,7 @@
return sOLT, nil
}
-func (a *adapter) setupRequestHandler(coreInstanceId string, iadapter adapters.IAdapter, coreProxy *com.CoreProxy) error {
+func (a *adapter) setupRequestHandler(ctx context.Context, coreInstanceId string, iadapter adapters.IAdapter, coreProxy *com.CoreProxy) error {
log.Info("setting-request-handler")
requestProxy := com.NewRequestHandlerProxy(coreInstanceId, iadapter, coreProxy)
if err := a.kip.SubscribeWithRequestHandlerInterface(kafka.Topic{Name: a.config.Topic}, requestProxy); err != nil {
@@ -227,11 +251,12 @@
return err
}
+ probe.UpdateStatusFromContext(ctx, "core-request-handler", probe.ServiceStatusRunning)
log.Info("request-handler-setup-done")
return nil
}
-func (a *adapter) registerWithCore(retries int) error {
+func (a *adapter) registerWithCore(ctx context.Context, retries int) error {
log.Info("registering-with-core")
adapterDescription := &voltha.Adapter{
Id: "simulated_olt",
@@ -254,6 +279,7 @@
break
}
}
+ probe.UpdateStatusFromContext(ctx, "register-with-core", probe.ServiceStatusRunning)
log.Info("registered-with-core")
return nil
}
@@ -329,7 +355,13 @@
defer cancel()
ad := newAdapter(cf)
- go ad.start(ctx)
+
+ p := &probe.Probe{}
+ go p.ListenAndServe(fmt.Sprintf("%s:%d", ad.config.ProbeHost, ad.config.ProbePort))
+
+ probeCtx := context.WithValue(ctx, probe.ProbeContextKey, p)
+
+ go ad.start(probeCtx)
code := waitForExit()
log.Infow("received-a-closing-signal", log.Fields{"code": code})
diff --git a/go.mod b/go.mod
index c4410a9..8846fd2 100644
--- a/go.mod
+++ b/go.mod
@@ -18,7 +18,7 @@
github.com/mitchellh/go-homedir v1.1.0 // indirect
github.com/mitchellh/go-testing-interface v1.0.0 // indirect
github.com/mitchellh/mapstructure v1.1.2 // indirect
- github.com/opencord/voltha-go v2.2.0+incompatible
+ github.com/opencord/voltha-go v2.2.1-0.20191011073237-626b2168bfbe+incompatible
github.com/opencord/voltha-protos v1.0.3
github.com/stretchr/testify v1.4.0
go.etcd.io/etcd v0.0.0-20190322183551-7a5acb4a43aa // indirect
diff --git a/go.sum b/go.sum
index dc97616..142b91c 100644
--- a/go.sum
+++ b/go.sum
@@ -132,8 +132,8 @@
github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
github.com/onsi/gomega v1.4.2 h1:3mYCb7aPxS/RU7TI1y4rkEn1oKmPRjNJLNEXgw7MH2I=
github.com/onsi/gomega v1.4.2/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
-github.com/opencord/voltha-go v2.2.0+incompatible h1:aaAKICXEGgVUnbRKOnTZfGPif2D2UailyDCDWinC52c=
-github.com/opencord/voltha-go v2.2.0+incompatible/go.mod h1:Zj45b2Bqq23E6sgXCnwonqfNOgRvOxN6tzKTYbatD1s=
+github.com/opencord/voltha-go v2.2.1-0.20191011073237-626b2168bfbe+incompatible h1:86RbNdgrk00NjoXoWQnWXPDelikiL8+iAqIqWkvpWkc=
+github.com/opencord/voltha-go v2.2.1-0.20191011073237-626b2168bfbe+incompatible/go.mod h1:Zj45b2Bqq23E6sgXCnwonqfNOgRvOxN6tzKTYbatD1s=
github.com/opencord/voltha-protos v1.0.3 h1:9v+R/QGF1xK+HKTqFM0IqCABoGCAxC8iKH4VzNBJDto=
github.com/opencord/voltha-protos v1.0.3/go.mod h1:myfFIkJdA+rCXmKdLImhh79MfabN4ZOKQ4grk32DnPQ=
github.com/pascaldekloe/goe v0.0.0-20180627143212-57f6aae5913c h1:Lgl0gzECD8GnQ5QCWA8o6BtfL6mDH5rQgM4/fX3avOs=
diff --git a/internal/pkg/config/config.go b/internal/pkg/config/config.go
index 5cf0dc0..57280a7 100644
--- a/internal/pkg/config/config.go
+++ b/internal/pkg/config/config.go
@@ -39,6 +39,8 @@
default_Topic = "simulated_olt"
default_CoreTopic = "rwcore"
default_OnuNumber = 1
+ default_ProbeHost = ""
+ default_ProbePort = 8080
)
// AdapterFlags represents the set of configurations used by the read-write adaptercore service
@@ -58,6 +60,8 @@
LogLevel int
OnuNumber int
Banner bool
+ ProbeHost string
+ ProbePort int
}
func init() {
@@ -81,6 +85,8 @@
LogLevel: default_LogLevel,
OnuNumber: default_OnuNumber,
Banner: default_Banner,
+ ProbeHost: default_ProbeHost,
+ ProbePort: default_ProbePort,
}
return &adapterFlags
}
@@ -129,6 +135,12 @@
help = fmt.Sprintf("Show startup banner log lines")
flag.BoolVar(&so.Banner, "banner", default_Banner, help)
+ help = fmt.Sprintf("The address on which to listen to answer liveness and readiness probe queries over HTTP.")
+ flag.StringVar(&(so.ProbeHost), "probe_host", default_ProbeHost, help)
+
+ help = fmt.Sprintf("The port on which to listen to answer liveness and readiness probe queries over HTTP.")
+ flag.IntVar(&(so.ProbePort), "probe_port", default_ProbePort, help)
+
flag.Parse()
containerName := getContainerInfo()
diff --git a/vendor/github.com/opencord/voltha-go/common/probe/probe.go b/vendor/github.com/opencord/voltha-go/common/probe/probe.go
new file mode 100644
index 0000000..8a8e485
--- /dev/null
+++ b/vendor/github.com/opencord/voltha-go/common/probe/probe.go
@@ -0,0 +1,253 @@
+/*
+ * Copyright 2019-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 probe
+
+import (
+ "context"
+ "fmt"
+ "github.com/opencord/voltha-go/common/log"
+ "net/http"
+ "sync"
+)
+
+// ProbeContextKey used to fetch the Probe instance from a context
+type ProbeContextKeyType string
+
+// ServiceStatus typed values for service status
+type ServiceStatus int
+
+const (
+ // ServiceStatusUnknown initial state of services
+ ServiceStatusUnknown ServiceStatus = iota
+
+ // ServiceStatusPreparing to optionally be used for prep, such as connecting
+ ServiceStatusPreparing
+
+ // ServiceStatusPrepared to optionally be used when prep is complete, but before run
+ ServiceStatusPrepared
+
+ // ServiceStatusRunning service is functional
+ ServiceStatusRunning
+
+ // ServiceStatusStopped service has stopped, but not because of error
+ ServiceStatusStopped
+
+ // ServiceStatusFailed service has stopped because of an error
+ ServiceStatusFailed
+)
+
+const (
+ // ProbeContextKey value of context key to fetch probe
+ ProbeContextKey = ProbeContextKeyType("status-update-probe")
+)
+
+// String convert ServiceStatus values to strings
+func (s ServiceStatus) String() string {
+ switch s {
+ default:
+ fallthrough
+ case ServiceStatusUnknown:
+ return "Unknown"
+ case ServiceStatusPreparing:
+ return "Preparing"
+ case ServiceStatusPrepared:
+ return "Prepared"
+ case ServiceStatusRunning:
+ return "Running"
+ case ServiceStatusStopped:
+ return "Stopped"
+ case ServiceStatusFailed:
+ return "Failed"
+ }
+}
+
+// ServiceStatusUpdate status update event
+type ServiceStatusUpdate struct {
+ Name string
+ Status ServiceStatus
+}
+
+// Probe reciever on which to implement probe capabilities
+type Probe struct {
+ readyFunc func(map[string]ServiceStatus) bool
+ healthFunc func(map[string]ServiceStatus) bool
+
+ mutex sync.RWMutex
+ status map[string]ServiceStatus
+ isReady bool
+ isHealthy bool
+}
+
+// WithReadyFunc override the default ready calculation function
+func (p *Probe) WithReadyFunc(readyFunc func(map[string]ServiceStatus) bool) *Probe {
+ p.readyFunc = readyFunc
+ return p
+}
+
+// WithHealthFunc override the default health calculation function
+func (p *Probe) WithHealthFunc(healthFunc func(map[string]ServiceStatus) bool) *Probe {
+ p.healthFunc = healthFunc
+ return p
+}
+
+// RegisterService register one or more service names with the probe, status will be track against service name
+func (p *Probe) RegisterService(names ...string) {
+ p.mutex.Lock()
+ defer p.mutex.Unlock()
+ if p.status == nil {
+ p.status = make(map[string]ServiceStatus)
+ }
+ for _, name := range names {
+ if _, ok := p.status[name]; !ok {
+ p.status[name] = ServiceStatusUnknown
+ log.Debugw("probe-service-registered", log.Fields{"service-name": name})
+ }
+ }
+
+ if p.readyFunc != nil {
+ p.isReady = p.readyFunc(p.status)
+ } else {
+ p.isReady = defaultReadyFunc(p.status)
+ }
+
+ if p.healthFunc != nil {
+ p.isHealthy = p.healthFunc(p.status)
+ } else {
+ p.isHealthy = defaultHealthFunc(p.status)
+ }
+}
+
+// UpdateStatus utility function to send a service update to the probe
+func (p *Probe) UpdateStatus(name string, status ServiceStatus) {
+ p.mutex.Lock()
+ defer p.mutex.Unlock()
+ if p.status == nil {
+ p.status = make(map[string]ServiceStatus)
+ }
+ p.status[name] = status
+ if p.readyFunc != nil {
+ p.isReady = p.readyFunc(p.status)
+ } else {
+ p.isReady = defaultReadyFunc(p.status)
+ }
+
+ if p.healthFunc != nil {
+ p.isHealthy = p.healthFunc(p.status)
+ } else {
+ p.isHealthy = defaultHealthFunc(p.status)
+ }
+ log.Debugw("probe-service-status-updated",
+ log.Fields{
+ "service-name": name,
+ "status": status.String(),
+ "ready": p.isReady,
+ "health": p.isHealthy,
+ })
+}
+
+// UpdateStatusFromContext a convenience function to pull the Probe reference from the
+// Context, if it exists, and then calling UpdateStatus on that Probe reference. If Context
+// is nil or if a Probe reference is not associated with the ProbeContextKey then nothing
+// happens
+func UpdateStatusFromContext(ctx context.Context, name string, status ServiceStatus) {
+ if ctx != nil {
+ if value := ctx.Value(ProbeContextKey); value != nil {
+ if p, ok := value.(*Probe); ok {
+ p.UpdateStatus(name, status)
+ }
+ }
+ }
+}
+
+// pulled out to a function to help better enable unit testing
+func (p *Probe) readzFunc(w http.ResponseWriter, req *http.Request) {
+ p.mutex.RLock()
+ defer p.mutex.RUnlock()
+ if p.isReady {
+ w.WriteHeader(http.StatusOK)
+ } else {
+ w.WriteHeader(http.StatusTeapot)
+ }
+}
+func (p *Probe) healthzFunc(w http.ResponseWriter, req *http.Request) {
+ p.mutex.RLock()
+ defer p.mutex.RUnlock()
+ if p.isHealthy {
+ w.WriteHeader(http.StatusOK)
+ } else {
+ w.WriteHeader(http.StatusTeapot)
+ }
+}
+func (p *Probe) detailzFunc(w http.ResponseWriter, req *http.Request) {
+ p.mutex.RLock()
+ defer p.mutex.RUnlock()
+ w.Header().Set("Content-Type", "application/json")
+ w.Write([]byte("{"))
+ comma := ""
+ for c, s := range p.status {
+ w.Write([]byte(fmt.Sprintf("%s\"%s\": \"%s\"", comma, c, s.String())))
+ comma = ", "
+ }
+ w.Write([]byte("}"))
+ w.WriteHeader(http.StatusOK)
+
+}
+
+// ListenAndServe implements 3 HTTP endpoints on the given port for healthz, readz, and detailz. Returns only on error
+func (p *Probe) ListenAndServe(address string) {
+ mux := http.NewServeMux()
+
+ // Returns the result of the readyFunc calculation
+ mux.HandleFunc("/readz", p.readzFunc)
+
+ // Returns the result of the healthFunc calculation
+ mux.HandleFunc("/healthz", p.healthzFunc)
+
+ // Returns the details of the services and their status as JSON
+ mux.HandleFunc("/detailz", p.detailzFunc)
+ s := &http.Server{
+ Addr: address,
+ Handler: mux,
+ }
+ log.Fatal(s.ListenAndServe())
+}
+
+// defaultReadyFunc if all services are running then ready, else not
+func defaultReadyFunc(services map[string]ServiceStatus) bool {
+ if len(services) == 0 {
+ return false
+ }
+ for _, status := range services {
+ if status != ServiceStatusRunning {
+ return false
+ }
+ }
+ return true
+}
+
+// defaultHealthFunc if no service is stopped or failed, then healthy, else not.
+// service is start as unknown, so they are considered healthy
+func defaultHealthFunc(services map[string]ServiceStatus) bool {
+ if len(services) == 0 {
+ return false
+ }
+ for _, status := range services {
+ if status == ServiceStatusStopped || status == ServiceStatusFailed {
+ return false
+ }
+ }
+ return true
+}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index fa6ee02..f03297a 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -50,11 +50,12 @@
github.com/mitchellh/go-homedir
# github.com/mitchellh/mapstructure v1.1.2
github.com/mitchellh/mapstructure
-# github.com/opencord/voltha-go v2.2.0+incompatible
+# github.com/opencord/voltha-go v2.2.1-0.20191011073237-626b2168bfbe+incompatible
github.com/opencord/voltha-go/adapters
github.com/opencord/voltha-go/adapters/common
github.com/opencord/voltha-go/common/log
github.com/opencord/voltha-go/common/version
+github.com/opencord/voltha-go/common/probe
github.com/opencord/voltha-go/db/kvstore
github.com/opencord/voltha-go/kafka
github.com/opencord/voltha-go/adapters/adapterif