Merge pull request #174 from datawire/lomb/issue143

implement some very basic ready and health probes

cmd/amb-sidecar/filters/app/app.go

@@ -4,6 +4,8 @@ import (
 	"net/http"
 	"strings"
 
+	"github.com/datawire/apro/cmd/amb-sidecar/filters/app/health"
+
 	"github.com/pkg/errors"
 	"golang.org/x/net/http2"
 	"golang.org/x/net/http2/h2c"
@@ -35,6 +37,15 @@ func NewFilterMux(config types.Config, logger types.Logger, controller *controll
 	grpcServer := grpc.NewServer()
 	filterapi.RegisterFilterService(grpcServer, filterMux)
 
+	// register more health probes off the MultiProbe as
+	probe := health.MultiProbe{
+		Logger: logger,
+	}
+
+	// this is a probe that always returns true... it is admittedly "dumb", but if the HTTP server stops serving
+	// this will fail and it forms the basis of the Probe API which we can use for subsequent more involved probes.
+	probe.RegisterProbe("basic", &health.StaticProbe{Value: true})
+
 	// The net/http.Server doesn't support h2c (unencrypted
 	// HTTP/2) built-in.  Since we want to have gRPC and plain
 	// HTTP/1 on the same unencrypted port, need h2c.
@@ -43,7 +54,19 @@ func NewFilterMux(config types.Config, logger types.Logger, controller *controll
 		if r.ProtoMajor == 2 && strings.HasPrefix(r.Header.Get("Content-Type"), "application/grpc") {
 			grpcServer.ServeHTTP(w, r)
 		} else {
-			http.NotFound(w, r)
+			path := r.URL.Path
+
+			// reserve the /_/sys/* path for future internal "system" paths.
+			if strings.HasPrefix(path, "/_/sys/healthz") || strings.HasPrefix(path, "/_/sys/readyz") {
+				healthy := probe.Check()
+				if healthy {
+					w.WriteHeader(http.StatusOK)
+				} else {
+					w.WriteHeader(http.StatusInternalServerError)
+				}
+			} else {
+				http.NotFound(w, r)
+			}
 		}
 	}), &http2.Server{}), nil
 }

cmd/amb-sidecar/filters/app/health/handlers.go

@@ -0,0 +1,94 @@
+package health
+
+import (
+	"github.com/datawire/apro/cmd/amb-sidecar/types"
+)
+
+type Probe interface {
+	Check() bool
+}
+
+// StaticProbe always returns the specified value. This is primarily for tests but has some limited value in development
+// scenarios.
+type StaticProbe struct {
+	Value bool
+}
+
+func (p *StaticProbe) Check() bool {
+	return p.Value
+}
+
+// RandomProbe returns a randomly selected boolean value. This is primarily for tests and development but has some
+// limited value in development.
+//type RandomProbe struct {
+//	rand *rand.Rand
+//}
+//
+//func (p *RandomProbe) Check() bool {
+//	return p.rand.Intn(2) == 0
+//}
+
+// MultiProbe executes zero or more probes.
+type MultiProbe struct {
+	Logger types.Logger
+	probes map[string]Probe
+}
+
+func (p *MultiProbe) RegisterProbe(name string, probe Probe) {
+	if p.probes == nil {
+		p.probes = make(map[string]Probe)
+	}
+
+	p.probes[name] = probe
+}
+
+func (p *MultiProbe) Check() bool {
+	if len(p.probes) == 0 {
+		p.Logger.Warn("no probes registered, assuming healthy")
+		return true
+	}
+
+	healthy := true
+	for name, probe := range p.probes {
+		probeResult := probe.Check()
+		l := p.Logger.
+			WithField("name", name).
+			WithField("result", probeResult)
+
+		if !probeResult {
+			l.Errorln("probe failed")
+			healthy = false
+			break
+		}
+
+		l.Info("probe succeeded")
+	}
+
+	return healthy
+}
+
+// SyncProbeHandler runs probes on demand as the handler is invoked.
+//
+// NOTE: An alternative implementation is to perform the checks asynchronously and just return a pre-computed result
+//       when the handler is invoked. That strategy is often employed when probes are expensive/slow and blocking
+//       the probing mechanism (e.g. Kubernetes liveness or readiness probes) would lead to failures.
+//type SyncProbeHandler struct {
+//	HealthinessProbe Probe
+//	ReadinessProbe   Probe
+//}
+//
+//func (h *SyncProbeHandler) QueryHealthiness(w http.ResponseWriter, r *http.Request) {
+//	if h.HealthinessProbe.Check() {
+//		w.WriteHeader(http.StatusOK)
+//	} else {
+//		w.WriteHeader(http.StatusInternalServerError)
+//	}
+//}
+//
+//func (h *SyncProbeHandler) QueryReadiness(w http.ResponseWriter, r *http.Request) {
+//	if h.ReadinessProbe.Check() {
+//		w.WriteHeader(http.StatusOK)
+//	} else {
+//		w.WriteHeader(http.StatusInternalServerError)
+//	}
+//}

k8s-sidecar/03-ambassador-pro-sidecar.yaml

@@ -304,12 +304,24 @@ spec:
       - name: ambassador-pro
         image: {{env "AMB_SIDECAR_IMAGE"}}
         ports:
-        - name: grpc-auth
-          containerPort: 8500
-        - name: grpc-ratelimit
-          containerPort: 8501
-        - name: http-debug
-          containerPort: 8502
+          - name: grpc-auth
+            containerPort: 8500
+          - name: grpc-ratelimit
+            containerPort: 8501
+          - name: http-debug
+            containerPort: 8502
+        livenessProbe:
+          httpGet:
+            path: /_/sys/healthz
+            port: 8500
+          initialDelaySeconds: 30
+          periodSeconds: 5
+        readinessProbe:
+          httpGet:
+            path: /_/sys/readyz
+            port: 8500
+          initialDelaySeconds: 30
+          periodSeconds: 5
         env:
         - name: REDIS_SOCKET_TYPE # For ratelimit
           value: tcp

k8s-standalone/03-ambassador-pro-standalone.yaml

@@ -213,6 +213,18 @@ spec:
       containers:
       - name: ambassador-pro
         image: {{env "AMB_SIDECAR_IMAGE"}}
+        livenessProbe:
+          httpGet:
+            path: /_/sys/healthz
+            port: 8500
+          initialDelaySeconds: 30
+          periodSeconds: 5
+        readinessProbe:
+          httpGet:
+            path: /_/sys/readyz
+            port: 8500
+          initialDelaySeconds: 30
+          periodSeconds: 5
         ports:
         - name: grpc-auth
           containerPort: 8500