feat(probe): add Kubernetes probe support with liveness, readiness, and startup checks (#3213)

* feat(probe): add Kubernetes probe support with liveness, readiness, and startup checks

Author: Alanxtl

common/constant/key.go

@@ -476,6 +476,13 @@ const (
 	PrometheusPushgatewayPasswordKey     = "prometheus.pushgateway.password"
 	PrometheusPushgatewayPushIntervalKey = "prometheus.pushgateway.push.interval"
 	PrometheusPushgatewayJobKey          = "prometheus.pushgateway.job"
+
+	ProbeEnabledKey          = "probe.enabled"
+	ProbePortKey             = "probe.port"
+	ProbeLivenessPathKey     = "probe.liveness.path"
+	ProbeReadinessPathKey    = "probe.readiness.path"
+	ProbeStartupPathKey      = "probe.startup.path"
+	ProbeUseInternalStateKey = "probe.use-internal-state"
 )
 
 // default meta cache config

common/constant/metric.go

@@ -17,6 +17,10 @@
 
 package constant
 
+import (
+	"time"
+)
+
 // metrics type
 const (
 	MetricsRegistry     = "dubbo.metrics.registry"
@@ -53,6 +57,13 @@ const (
 	PrometheusDefaultMetricsPort        = "9090"
 	PrometheusDefaultPushInterval       = 30
 	PrometheusDefaultJobName            = "default_dubbo_job"
+	ProbeDefaultPort                    = "22222"
+	ProbeDefaultLivenessPath            = "/live"
+	ProbeDefaultReadinessPath           = "/ready"
+	ProbeDefaultStartupPath             = "/startup"
+	ProbeReadHeaderTimeout              = 5 * time.Second
+	ProbeWriteTimeout                   = 10 * time.Second
+	ProbeIdleTimeout                    = 30 * time.Second
 	MetricFilterStartTime               = "metric_filter_start_time"
 )
 

compat.go

@@ -379,6 +379,21 @@ func compatMetricConfig(c *global.MetricsConfig) *config.MetricsConfig {
 		EnableMetadata:     c.EnableMetadata,
 		EnableRegistry:     c.EnableRegistry,
 		EnableConfigCenter: c.EnableConfigCenter,
+		Probe:              compatMetricProbeConfig(c.Probe),
+	}
+}
+
+func compatMetricProbeConfig(c *global.ProbeConfig) *config.ProbeConfig {
+	if c == nil {
+		return nil
+	}
+	return &config.ProbeConfig{
+		Enabled:          c.Enabled,
+		Port:             c.Port,
+		LivenessPath:     c.LivenessPath,
+		ReadinessPath:    c.ReadinessPath,
+		StartupPath:      c.StartupPath,
+		UseInternalState: c.UseInternalState,
 	}
 }
 
@@ -916,6 +931,21 @@ func compatGlobalMetricConfig(c *config.MetricsConfig) *global.MetricsConfig {
 		EnableMetadata:     c.EnableMetadata,
 		EnableRegistry:     c.EnableRegistry,
 		EnableConfigCenter: c.EnableConfigCenter,
+		Probe:              compatGlobalMetricProbeConfig(c.Probe),
+	}
+}
+
+func compatGlobalMetricProbeConfig(c *config.ProbeConfig) *global.ProbeConfig {
+	if c == nil {
+		return nil
+	}
+	return &global.ProbeConfig{
+		Enabled:          c.Enabled,
+		Port:             c.Port,
+		LivenessPath:     c.LivenessPath,
+		ReadinessPath:    c.ReadinessPath,
+		StartupPath:      c.StartupPath,
+		UseInternalState: c.UseInternalState,
 	}
 }
 

config/metric_config.go

@@ -44,6 +44,7 @@ type MetricsConfig struct {
 	EnableConfigCenter *bool             `default:"false" yaml:"enable-config-center" json:"enable-config-center,omitempty" property:"enable-config-center"`
 	Prometheus         *PrometheusConfig `yaml:"prometheus" json:"prometheus" property:"prometheus"`
 	Aggregation        *AggregateConfig  `yaml:"aggregation" json:"aggregation" property:"aggregation"`
+	Probe              *ProbeConfig      `yaml:"probe" json:"probe" property:"probe"`
 	rootConfig         *RootConfig
 }
 
@@ -58,6 +59,15 @@ type PrometheusConfig struct {
 	Pushgateway *PushgatewayConfig `yaml:"pushgateway" json:"pushgateway,omitempty" property:"pushgateway"`
 }
 
+type ProbeConfig struct {
+	Enabled          *bool  `default:"false" yaml:"enabled" json:"enabled,omitempty" property:"enabled"`
+	Port             string `default:"22222" yaml:"port" json:"port,omitempty" property:"port"`
+	LivenessPath     string `default:"/live" yaml:"liveness-path" json:"liveness-path,omitempty" property:"liveness-path"`
+	ReadinessPath    string `default:"/ready" yaml:"readiness-path" json:"readiness-path,omitempty" property:"readiness-path"`
+	StartupPath      string `default:"/startup" yaml:"startup-path" json:"startup-path,omitempty" property:"startup-path"`
+	UseInternalState *bool  `default:"true" yaml:"use-internal-state" json:"use-internal-state,omitempty" property:"use-internal-state"`
+}
+
 type Exporter struct {
 	Enabled *bool `default:"true" yaml:"enabled" json:"enabled,omitempty" property:"enabled"`
 }

global/metric_config.go

@@ -28,6 +28,7 @@ type MetricsConfig struct {
 	EnableMetadata     *bool             `default:"true" yaml:"enable-metadata" json:"enable-metadata,omitempty" property:"enable-metadata"`
 	EnableRegistry     *bool             `default:"true" yaml:"enable-registry" json:"enable-registry,omitempty" property:"enable-registry"`
 	EnableConfigCenter *bool             `default:"true" yaml:"enable-config-center" json:"enable-config-center,omitempty" property:"enable-config-center"`
+	Probe              *ProbeConfig      `yaml:"probe" json:"probe" property:"probe"`
 }
 
 type AggregateConfig struct {
@@ -41,6 +42,15 @@ type PrometheusConfig struct {
 	Pushgateway *PushgatewayConfig `yaml:"pushgateway" json:"pushgateway,omitempty" property:"pushgateway"`
 }
 
+type ProbeConfig struct {
+	Enabled          *bool  `default:"false" yaml:"enabled" json:"enabled,omitempty" property:"enabled"`
+	Port             string `default:"22222" yaml:"port" json:"port,omitempty" property:"port"`
+	LivenessPath     string `default:"/live" yaml:"liveness-path" json:"liveness-path,omitempty" property:"liveness-path"`
+	ReadinessPath    string `default:"/ready" yaml:"readiness-path" json:"readiness-path,omitempty" property:"readiness-path"`
+	StartupPath      string `default:"/startup" yaml:"startup-path" json:"startup-path,omitempty" property:"startup-path"`
+	UseInternalState *bool  `default:"true" yaml:"use-internal-state" json:"use-internal-state,omitempty" property:"use-internal-state"`
+}
+
 type Exporter struct {
 	Enabled *bool `default:"false" yaml:"enabled" json:"enabled,omitempty" property:"enabled"`
 }
@@ -57,7 +67,7 @@ type PushgatewayConfig struct {
 
 func DefaultMetricsConfig() *MetricsConfig {
 	// return a new config without setting any field means there is not any default value for initialization
-	return &MetricsConfig{Prometheus: defaultPrometheusConfig(), Aggregation: defaultAggregateConfig()}
+	return &MetricsConfig{Prometheus: defaultPrometheusConfig(), Aggregation: defaultAggregateConfig(), Probe: defaultProbeConfig()}
 }
 
 // Clone a new MetricsConfig
@@ -100,6 +110,7 @@ func (c *MetricsConfig) Clone() *MetricsConfig {
 		EnableMetadata:     newEnableMetadata,
 		EnableRegistry:     newEnableRegistry,
 		EnableConfigCenter: newEnableConfigCenter,
+		Probe:              c.Probe.Clone(),
 	}
 }
 
@@ -155,6 +166,36 @@ func (c *PushgatewayConfig) Clone() *PushgatewayConfig {
 	}
 }
 
+func defaultProbeConfig() *ProbeConfig {
+	return &ProbeConfig{}
+}
+
+func (c *ProbeConfig) Clone() *ProbeConfig {
+	if c == nil {
+		return nil
+	}
+
+	var newEnabled *bool
+	if c.Enabled != nil {
+		newEnabled = new(bool)
+		*newEnabled = *c.Enabled
+	}
+	var newUseInternalState *bool
+	if c.UseInternalState != nil {
+		newUseInternalState = new(bool)
+		*newUseInternalState = *c.UseInternalState
+	}
+
+	return &ProbeConfig{
+		Enabled:          newEnabled,
+		Port:             c.Port,
+		LivenessPath:     c.LivenessPath,
+		ReadinessPath:    c.ReadinessPath,
+		StartupPath:      c.StartupPath,
+		UseInternalState: newUseInternalState,
+	}
+}
+
 func defaultAggregateConfig() *AggregateConfig {
 	return &AggregateConfig{}
 }

metrics/options.go

@@ -152,3 +152,41 @@ func WithPath(path string) Option {
 		opts.Metrics.Path = path
 	}
 }
+
+// Below are options for probe
+func WithProbeEnabled() Option {
+	return func(opts *Options) {
+		b := true
+		opts.Metrics.Probe.Enabled = &b
+	}
+}
+
+func WithProbePort(port int) Option {
+	return func(opts *Options) {
+		opts.Metrics.Probe.Port = strconv.Itoa(port)
+	}
+}
+
+func WithProbeLivenessPath(path string) Option {
+	return func(opts *Options) {
+		opts.Metrics.Probe.LivenessPath = path
+	}
+}
+
+func WithProbeReadinessPath(path string) Option {
+	return func(opts *Options) {
+		opts.Metrics.Probe.ReadinessPath = path
+	}
+}
+
+func WithProbeStartupPath(path string) Option {
+	return func(opts *Options) {
+		opts.Metrics.Probe.StartupPath = path
+	}
+}
+
+func WithProbeUseInternalState(use bool) Option {
+	return func(opts *Options) {
+		opts.Metrics.Probe.UseInternalState = &use
+	}
+}

metrics/probe/http.go

@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 (
+	"errors"
+	"net/http"
+)
+
+var (
+	errNotReady   = errors.New("not ready")
+	errNotStarted = errors.New("not started")
+)
+
+func livenessHandler(w http.ResponseWriter, r *http.Request) {
+	// k8s only use GET
+	if r.Method != http.MethodGet {
+		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
+		return
+	}
+
+	if err := CheckLiveness(r.Context()); err != nil {
+		http.Error(w, err.Error(), http.StatusServiceUnavailable)
+		return
+	}
+	w.WriteHeader(http.StatusOK)
+	_, _ = w.Write([]byte("ok"))
+}
+
+func readinessHandler(w http.ResponseWriter, r *http.Request) {
+	// k8s only use GET
+	if r.Method != http.MethodGet {
+		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
+		return
+	}
+
+	if err := CheckReadiness(r.Context()); err != nil {
+		http.Error(w, err.Error(), http.StatusServiceUnavailable)
+		return
+	}
+	w.WriteHeader(http.StatusOK)
+	_, _ = w.Write([]byte("ok"))
+}
+
+func startupHandler(w http.ResponseWriter, r *http.Request) {
+	// k8s only use GET
+	if r.Method != http.MethodGet {
+		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
+		return
+	}
+
+	if err := CheckStartup(r.Context()); err != nil {
+		http.Error(w, err.Error(), http.StatusServiceUnavailable)
+		return
+	}
+	w.WriteHeader(http.StatusOK)
+	_, _ = w.Write([]byte("ok"))
+}

metrics/probe/http_test.go

@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+)
+
+func TestHandlersSuccess(t *testing.T) {
+	resetProbeState()
+
+	RegisterLiveness("ok", func(context.Context) error { return nil })
+	RegisterReadiness("ok", func(context.Context) error { return nil })
+	RegisterStartup("ok", func(context.Context) error { return nil })
+
+	req := httptest.NewRequest(http.MethodGet, "/live", nil)
+	rec := httptest.NewRecorder()
+	livenessHandler(rec, req)
+	if rec.Code != http.StatusOK {
+		t.Fatalf("expected 200 for liveness, got %d", rec.Code)
+	}
+
+	req = httptest.NewRequest(http.MethodGet, "/ready", nil)
+	rec = httptest.NewRecorder()
+	readinessHandler(rec, req)
+	if rec.Code != http.StatusOK {
+		t.Fatalf("expected 200 for readiness, got %d", rec.Code)
+	}
+
+	req = httptest.NewRequest(http.MethodGet, "/startup", nil)
+	rec = httptest.NewRecorder()
+	startupHandler(rec, req)
+	if rec.Code != http.StatusOK {
+		t.Fatalf("expected 200 for startup, got %d", rec.Code)
+	}
+}
+
+func TestHandlersFailure(t *testing.T) {
+	resetProbeState()
+
+	RegisterLiveness("fail", func(context.Context) error { return errors.New("bad") })
+	RegisterReadiness("fail", func(context.Context) error { return errors.New("bad") })
+	RegisterStartup("fail", func(context.Context) error { return errors.New("bad") })
+
+	req := httptest.NewRequest(http.MethodGet, "/live", nil)
+	rec := httptest.NewRecorder()
+	livenessHandler(rec, req)
+	if rec.Code != http.StatusServiceUnavailable {
+		t.Fatalf("expected 503 for liveness, got %d", rec.Code)
+	}
+
+	req = httptest.NewRequest(http.MethodGet, "/ready", nil)
+	rec = httptest.NewRecorder()
+	readinessHandler(rec, req)
+	if rec.Code != http.StatusServiceUnavailable {
+		t.Fatalf("expected 503 for readiness, got %d", rec.Code)
+	}
+
+	req = httptest.NewRequest(http.MethodGet, "/startup", nil)
+	rec = httptest.NewRecorder()
+	startupHandler(rec, req)
+	if rec.Code != http.StatusServiceUnavailable {
+		t.Fatalf("expected 503 for startup, got %d", rec.Code)
+	}
+}