feat: add base metrics endpoint (#37)

Co-authored-by: lynx009 <2030509072@qq.com>

Author: yuluo-yx

etc/hertzbeat-collector.yaml

@@ -33,3 +33,6 @@ collector:
   # Collector identity and mode
   identity: hertzbeat-collector-go
   mode: public
+
+  metrics:
+    port: 9090

internal/cmd/server.go

@@ -31,6 +31,7 @@ import (
 	cfgloader "hertzbeat.apache.org/hertzbeat-collector-go/internal/config"
 	"hertzbeat.apache.org/hertzbeat-collector-go/internal/job/collect"
 	jobserver "hertzbeat.apache.org/hertzbeat-collector-go/internal/job/server"
+	"hertzbeat.apache.org/hertzbeat-collector-go/internal/metrics"
 	clrserver "hertzbeat.apache.org/hertzbeat-collector-go/internal/server"
 	transportserver "hertzbeat.apache.org/hertzbeat-collector-go/internal/transport"
 	collectortypes "hertzbeat.apache.org/hertzbeat-collector-go/internal/types/collector"
@@ -109,13 +110,15 @@ func server(ctx context.Context, logOut io.Writer) error {
 }
 
 func startRunners(ctx context.Context, cfg *clrserver.Server) error {
+
 	// Create transport server first
-	transportRunner := transportserver.NewFromConfig(cfg.Config)
+	transportRunner := transportserver.New(cfg)
 
 	// Create lazy message router that can get transport client when needed
 	messageRouter := collect.NewLazyMessageRouter(transportRunner, cfg.Logger, cfg.Config.Collector.Identity)
 
 	// Create job server with message router
+	// todo optimize not depend server start!
 	jobRunner := jobserver.New(&jobserver.Config{
 		Server:        *cfg,
 		MessageRouter: messageRouter,
@@ -124,12 +127,15 @@ func startRunners(ctx context.Context, cfg *clrserver.Server) error {
 	// Connect transport to job scheduler
 	transportRunner.SetJobScheduler(jobRunner)
 
+	// Create metrics runner
+	metricsRunner := metrics.New(cfg)
+
 	runners := []struct {
 		runner Runner[collectortypes.Info]
 	}{
+		{metricsRunner},
 		{jobRunner},
 		{transportRunner},
-		// todo; add metrics
 	}
 
 	errCh := make(chan error, len(runners))

internal/metrics/metrics.go

@@ -18,3 +18,137 @@
  */
 
 package metrics
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"net/http"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+	clrserver "hertzbeat.apache.org/hertzbeat-collector-go/internal/server"
+	"hertzbeat.apache.org/hertzbeat-collector-go/internal/types/collector"
+	"hertzbeat.apache.org/hertzbeat-collector-go/internal/util/logger"
+)
+
+const (
+	Namespace = "hertzbeat"
+	Subsystem = "collector"
+)
+
+var (
+	// JobExecutionTotal counts the total number of job executions
+	JobExecutionTotal = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Namespace: Namespace,
+			Subsystem: Subsystem,
+			Name:      "job_execution_total",
+			Help:      "Total number of job executions",
+		},
+		[]string{"status", "type"},
+	)
+
+	// JobExecutionDuration tracks the duration of job executions
+	JobExecutionDuration = prometheus.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Namespace: Namespace,
+			Subsystem: Subsystem,
+			Name:      "job_execution_duration_seconds",
+			Help:      "Duration of job executions in seconds",
+			Buckets:   prometheus.DefBuckets,
+		},
+		[]string{"type"},
+	)
+
+	// CollectorUp indicates if the collector is up
+	CollectorUp = prometheus.NewGauge(
+		prometheus.GaugeOpts{
+			Namespace: Namespace,
+			Subsystem: Subsystem,
+			Name:      "up",
+			Help:      "1 if the collector is up, 0 otherwise",
+		},
+	)
+)
+
+func init() {
+	// Register metrics with the global prometheus registry
+	prometheus.MustRegister(JobExecutionTotal)
+	prometheus.MustRegister(JobExecutionDuration)
+	prometheus.MustRegister(CollectorUp)
+	CollectorUp.Set(1)
+}
+
+// Runner implements the metrics server runner
+type Runner struct {
+	cfg    *clrserver.Server
+	server *http.Server
+}
+
+// New creates a new metrics runner
+func New(cfg *clrserver.Server) *Runner {
+
+	return &Runner{cfg: cfg}
+}
+
+// Start starts the metrics server
+func (r *Runner) Start(ctx context.Context) error {
+
+	// init logger
+	mlog := r.initLogs()
+
+	mux := http.NewServeMux()
+	mux.Handle("/metrics", promhttp.Handler())
+
+	addr := fmt.Sprintf(":%d", r.cfg.Config.Collector.MetricsConfig.Port)
+	r.server = &http.Server{
+		Addr:              addr,
+		Handler:           mux,
+		ReadTimeout:       5 * time.Second,
+		ReadHeaderTimeout: 5 * time.Second,
+		WriteTimeout:      10 * time.Second,
+		IdleTimeout:       15 * time.Second,
+		MaxHeaderBytes:    1 << 20, // 1 MB
+		BaseContext: func(_ net.Listener) context.Context {
+			return ctx
+		},
+	}
+
+	mlog.Info("Starting metrics server", "addr", addr)
+
+	go func() {
+		if err := r.server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
+			mlog.Error(err, "Metrics server failed")
+		}
+	}()
+
+	<-ctx.Done()
+	return r.Close()
+}
+
+// Info returns the runner info
+func (r *Runner) Info() collector.Info {
+
+	return collector.Info{
+		Name: "metrics-server",
+	}
+}
+
+// Close closes the metrics server
+func (r *Runner) Close() error {
+
+	if r.server != nil {
+		r.initLogs().Info("Shutting down metrics server")
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		return r.server.Shutdown(ctx)
+	}
+	return nil
+}
+
+func (r *Runner) initLogs() logger.Logger {
+
+	return r.cfg.Logger.WithName("metrics")
+}