CaaS - batch probe fixes (#2220)

Author: RobertLucian

cmd/dequeuer/main.go

@@ -87,6 +87,8 @@ func main() {
 		log.Fatal("--api-name is a required option")
 	case apiKind == "":
 		log.Fatal("--api-kind is a required option")
+	case adminPort == 0:
+		log.Fatal("--admin-port is a required option")
 	}
 
 	targetURL := "http://127.0.0.1:" + strconv.Itoa(userContainerPort)
@@ -142,8 +144,6 @@ func main() {
 	var dequeuerConfig dequeuer.SQSDequeuerConfig
 	var messageHandler dequeuer.MessageHandler
 
-	errCh := make(chan error)
-
 	switch apiKind {
 	case userconfig.BatchAPIKind.String():
 		if jobID == "" {
@@ -169,9 +169,6 @@ func main() {
 		if clusterUID == "" {
 			log.Fatal("--cluster-uid is a required option")
 		}
-		if adminPort == 0 {
-			log.Fatal("--admin-port is a required option")
-		}
 
 		config := dequeuer.AsyncMessageHandlerConfig{
 			ClusterUID: clusterUID,
@@ -187,24 +184,26 @@ func main() {
 			StopIfNoMessages: false,
 		}
 
-		adminHandler := http.NewServeMux()
-		adminHandler.Handle("/healthz", dequeuer.HealthcheckHandler(func() bool {
-			return probe.AreProbesHealthy(probes)
-		}))
-
-		go func() {
-			server := &http.Server{
-				Addr:    ":" + strconv.Itoa(adminPort),
-				Handler: adminHandler,
-			}
-			log.Infof("Starting %s server on %s", "admin", server.Addr)
-			errCh <- server.ListenAndServe()
-		}()
-
 	default:
 		exit(log, err, fmt.Sprintf("kind %s is not supported", apiKind))
 	}
 
+	errCh := make(chan error)
+
+	adminHandler := http.NewServeMux()
+	adminHandler.Handle("/healthz", dequeuer.HealthcheckHandler(func() bool {
+		return probe.AreProbesHealthy(probes)
+	}))
+
+	go func() {
+		server := &http.Server{
+			Addr:    ":" + strconv.Itoa(adminPort),
+			Handler: adminHandler,
+		}
+		log.Infof("Starting %s server on %s", "admin", server.Addr)
+		errCh <- server.ListenAndServe()
+	}()
+
 	sigint := make(chan os.Signal, 1)
 	signal.Notify(sigint, os.Interrupt)
 

pkg/probe/probe.go

@@ -173,7 +173,9 @@ func (p *Probe) httpProbe() error {
 		"http://",
 	)
 
-	httpClient := &http.Client{}
+	httpClient := &http.Client{
+		Timeout: time.Duration(p.TimeoutSeconds) * time.Second,
+	}
 	req, err := http.NewRequest(http.MethodGet, targetURL, nil)
 	if err != nil {
 		return err

pkg/workloads/k8s.go

@@ -152,7 +152,7 @@ func asyncDequeuerProxyContainer(api spec.API, queueURL string) (kcore.Container
 			},
 			InitialDelaySeconds: 1,
 			TimeoutSeconds:      1,
-			PeriodSeconds:       5,
+			PeriodSeconds:       10,
 			SuccessThreshold:    1,
 			FailureThreshold:    1,
 		},
@@ -180,6 +180,7 @@ func batchDequeuerProxyContainer(api spec.API, jobID, queueURL string) (kcore.Co
 			"--job-id", jobID,
 			"--user-port", s.Int32(*api.Pod.Port),
 			"--statsd-port", consts.StatsDPortStr,
+			"--admin-port", consts.AdminPortStr,
 		},
 		Env: append(baseEnvVars, kcore.EnvVar{
 			Name: "HOST_IP",
@@ -189,6 +190,19 @@ func batchDequeuerProxyContainer(api spec.API, jobID, queueURL string) (kcore.Co
 				},
 			},
 		}),
+		ReadinessProbe: &kcore.Probe{
+			Handler: kcore.Handler{
+				HTTPGet: &kcore.HTTPGetAction{
+					Path: "/healthz",
+					Port: intstr.FromInt(int(consts.AdminPortInt32)),
+				},
+			},
+			InitialDelaySeconds: 1,
+			TimeoutSeconds:      1,
+			PeriodSeconds:       10,
+			SuccessThreshold:    1,
+			FailureThreshold:    1,
+		},
 		VolumeMounts: []kcore.VolumeMount{
 			ClusterConfigMount(),
 			CortexMount(),
@@ -233,7 +247,7 @@ func realtimeProxyContainer(api spec.API) (kcore.Container, kcore.Volume) {
 			},
 			InitialDelaySeconds: 1,
 			TimeoutSeconds:      1,
-			PeriodSeconds:       5,
+			PeriodSeconds:       10,
 			SuccessThreshold:    1,
 			FailureThreshold:    1,
 		},