Merge pull request #108500 from SergeyKanzhelev/grpcsupportInAgnHost

add the grpc-health-checking to agnhost

test/images/agnhost/Dockerfile

@@ -38,7 +38,8 @@ RUN tar -xzvf /coredns.tgz && rm -f /coredns.tgz
 # PORT 8080 needed by: netexec, nettest, resource-consumer, resource-consumer-controller
 # PORT 8081 needed by: netexec
 # PORT 9376 needed by: serve-hostname
-EXPOSE 80 8080 8081 9376
+# PORT 5000 needed by: grpc-health-checking
+EXPOSE 80 8080 8081 9376 5000
 
 # from netexec
 RUN mkdir /uploads

test/images/agnhost/README.md

@@ -254,6 +254,28 @@ Usage:
     kubectl exec test-agnhost -- /agnhost liveness
 ```
 
+### grpc-health-checking
+
+Started the gRPC health checking server. The health checking response can be
+controlled with the time delay or via http control server.
+
+- `--delay-unhealthy-sec` - the delay to change status to NOT_SERVING.
+  Endpoint reporting SERVING for `delay-unhealthy-sec` (`-1` by default)
+  seconds and then NOT_SERVING. Negative value indicates always SERVING. Use `0` to
+  start endpoint as NOT_SERVING.
+- `--port` (default: `5000`) can be used to override the gRPC port number.
+- `--http-port` (default: `8080`) can be used to override the http control server port number.
+- `--service` (default: ``) can be used used to specify which service this endpoint will respond to.
+
+Usage:
+
+```console
+    kubectl exec test-agnhost -- /agnhost grpc-health-checking \
+      [--delay-unhealthy-sec 5] [--service ""] \
+      [--port 5000] [--http-port 8080]
+
+    kubectl exec test-agnhost -- curl http://localhost:8080/make-not-serving
+```
 
 ### logs-generator
 

test/images/agnhost/VERSION

@@ -1 +1 @@
-2.34
+2.35

test/images/agnhost/agnhost.go

@@ -28,6 +28,7 @@ import (
 	"k8s.io/kubernetes/test/images/agnhost/dns"
 	"k8s.io/kubernetes/test/images/agnhost/entrypoint-tester"
 	"k8s.io/kubernetes/test/images/agnhost/fakegitserver"
+	grpchealthchecking "k8s.io/kubernetes/test/images/agnhost/grpc-health-checking"
 	"k8s.io/kubernetes/test/images/agnhost/guestbook"
 	"k8s.io/kubernetes/test/images/agnhost/inclusterclient"
 	"k8s.io/kubernetes/test/images/agnhost/liveness"
@@ -82,6 +83,7 @@ func main() {
 	rootCmd.AddCommand(testwebserver.CmdTestWebserver)
 	rootCmd.AddCommand(webhook.CmdWebhook)
 	rootCmd.AddCommand(openidmetadata.CmdTestServiceAccountIssuerDiscovery)
+	rootCmd.AddCommand(grpchealthchecking.CmdGrpcHealthChecking)
 
 	// NOTE(claudiub): Some tests are passing logging related flags, so we need to be able to
 	// accept them. This will also include them in the printed help.

test/images/agnhost/grpc-health-checking/grpc-health-checking.go

@@ -0,0 +1,138 @@
+/*
+Copyright 2022 The Kubernetes Authors.
+
+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 grpchealthchecking offers a tiny grpc health checking endpoint.
+package grpchealthchecking
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"net"
+	"time"
+
+	"net/http"
+
+	"github.com/spf13/cobra"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/health/grpc_health_v1"
+	"google.golang.org/grpc/status"
+)
+
+// CmdGrpcHealthChecking is used by agnhost Cobra.
+var CmdGrpcHealthChecking = &cobra.Command{
+	Use:   "grpc-health-checking",
+	Short: "Starts a simple grpc health checking endpoint",
+	Long:  "Starts a simple grpc health checking endpoint with --port to serve on and --service to check status for. The endpoint returns SERVING for the first --delay-unhealthy-sec, and NOT_SERVING after this. NOT_FOUND will be returned for the requests for non-configured service name. Probe can be forced to be set NOT_SERVING by calling /make-not-serving http endpoint.",
+	Args:  cobra.MaximumNArgs(0),
+	Run:   main,
+}
+
+var (
+	port              int
+	httpPort          int
+	delayUnhealthySec int
+	service           string
+	forceUnhealthy    *bool
+)
+
+func init() {
+	CmdGrpcHealthChecking.Flags().IntVar(&port, "port", 5000, "Port number.")
+	CmdGrpcHealthChecking.Flags().IntVar(&httpPort, "http-port", 8080, "Port number for the /make-serving and /make-not-serving.")
+	CmdGrpcHealthChecking.Flags().IntVar(&delayUnhealthySec, "delay-unhealthy-sec", -1, "Number of seconds to delay before start reporting NOT_SERVING, negative value indicates never.")
+	CmdGrpcHealthChecking.Flags().StringVar(&service, "service", "", "Service name to register the health check for.")
+	forceUnhealthy = nil
+}
+
+type HealthChecker struct {
+	started time.Time
+}
+
+func (s *HealthChecker) Check(ctx context.Context, req *grpc_health_v1.HealthCheckRequest) (*grpc_health_v1.HealthCheckResponse, error) {
+	log.Printf("Serving the Check request for health check, started at %v", s.started)
+
+	if req.Service != service {
+		return nil, status.Errorf(codes.NotFound, "unknown service")
+	}
+
+	duration := time.Since(s.started)
+	if ((forceUnhealthy != nil) && *forceUnhealthy) || ((delayUnhealthySec >= 0) && (duration.Seconds() >= float64(delayUnhealthySec))) {
+		return &grpc_health_v1.HealthCheckResponse{
+			Status: grpc_health_v1.HealthCheckResponse_NOT_SERVING,
+		}, nil
+	}
+
+	return &grpc_health_v1.HealthCheckResponse{
+		Status: grpc_health_v1.HealthCheckResponse_SERVING,
+	}, nil
+}
+
+func (s *HealthChecker) Watch(req *grpc_health_v1.HealthCheckRequest, server grpc_health_v1.Health_WatchServer) error {
+	return status.Error(codes.Unimplemented, "unimplemented")
+}
+
+func NewHealthChecker(started time.Time) *HealthChecker {
+	return &HealthChecker{
+		started: started,
+	}
+}
+
+func main(cmd *cobra.Command, args []string) {
+	started := time.Now()
+
+	http.HandleFunc("/make-not-serving", func(w http.ResponseWriter, r *http.Request) {
+		log.Printf("Mark as unhealthy")
+		forceUnhealthy = new(bool)
+		*forceUnhealthy = true
+		w.WriteHeader(200)
+		data := (time.Since(started)).String()
+		w.Write([]byte(data))
+	})
+
+	http.HandleFunc("/make-serving", func(w http.ResponseWriter, r *http.Request) {
+		log.Printf("Mark as healthy")
+		forceUnhealthy = new(bool)
+		*forceUnhealthy = false
+		w.WriteHeader(200)
+		data := (time.Since(started)).String()
+		w.Write([]byte(data))
+	})
+
+	go func() {
+		httpServerAdr := fmt.Sprintf(":%d", httpPort)
+		log.Printf("Http server starting to listen on %s", httpServerAdr)
+		log.Fatal(http.ListenAndServe(httpServerAdr, nil))
+	}()
+
+	serverAdr := fmt.Sprintf(":%d", port)
+	listenAddr, err := net.Listen("tcp", serverAdr)
+	if err != nil {
+		log.Fatal(fmt.Sprintf("Error while starting the listening service %v", err.Error()))
+	}
+
+	grpcServer := grpc.NewServer()
+	healthService := NewHealthChecker(started)
+	grpc_health_v1.RegisterHealthServer(grpcServer, healthService)
+
+	log.Printf("gRPC server starting to listen on %s", serverAdr)
+	if err = grpcServer.Serve(listenAddr); err != nil {
+		log.Fatal(fmt.Sprintf("Error while starting the gRPC server on the %s listen address %v", listenAddr, err.Error()))
+	}
+
+	select {}
+}