diff --git a/pkg/controllers/cnset/controller.go b/pkg/controllers/cnset/controller.go index dd92db3e..b36965b8 100644 --- a/pkg/controllers/cnset/controller.go +++ b/pkg/controllers/cnset/controller.go @@ -16,6 +16,7 @@ package cnset import ( "fmt" + "strconv" "time" "github.com/matrixorigin/matrixone-operator/api/features" @@ -93,6 +94,10 @@ func (c *Actor) Observe(ctx *recon.Context[*v1alpha1.CNSet]) (recon.Action[*v1al return nil, errors.WrapPrefix(err, "sync service", 0) } + if err := c.syncMetricService(ctx); err != nil { + return nil, errors.WrapPrefix(err, "sync metric service", 0) + } + // diff desired cloneset and determine whether should an update be invoked origin := cs.DeepCopy() if err := syncCloneSet(ctx, cs); err != nil { @@ -184,6 +189,41 @@ func (c *Actor) Observe(ctx *recon.Context[*v1alpha1.CNSet]) (recon.Action[*v1al return nil, recon.ErrReSync("cnset is not ready or synced", reSyncAfter) } +// syncMetricService reconciles a dedicated ClusterIP Service exposing the CN metrics port, +// so that Service-based Prometheus discovery (e.g. ServiceMonitor matching on port name +// "metric") can find CN targets the same way it already does for DN/Log (issue #600). +func (c *Actor) syncMetricService(ctx *recon.Context[*v1alpha1.CNSet]) error { + cn := ctx.Obj + svc := &corev1.Service{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: cn.Namespace, + Name: metricSvcName(cn), + Labels: common.SubResourceLabels(cn), + }, + Spec: corev1.ServiceSpec{ + Selector: common.SubResourceLabels(cn), + }, + } + return recon.CreateOwnedOrUpdate(ctx, svc, func() error { + svc.Spec.Type = corev1.ServiceTypeClusterIP + svc.Spec.Ports = []corev1.ServicePort{{ + Name: "metric", + Port: int32(common.MetricsPort), + }} + if cn.Spec.PromDiscoveredByService() { + if svc.Annotations == nil { + svc.Annotations = map[string]string{} + } + svc.Annotations[common.PrometheusScrapeAnno] = "true" + svc.Annotations[common.PrometheusPortAnno] = strconv.Itoa(common.MetricsPort) + } else { + delete(svc.Annotations, common.PrometheusScrapeAnno) + delete(svc.Annotations, common.PrometheusPortAnno) + } + return nil + }) +} + func (c *WithResources) Scale(ctx *recon.Context[*v1alpha1.CNSet]) error { return ctx.Patch(c.cs, func() error { scaleSet(ctx.Obj, c.cs) @@ -208,6 +248,8 @@ func (c *Actor) Finalize(ctx *recon.Context[*v1alpha1.CNSet]) (bool, error) { Name: setName(cn), }}, &corev1.Service{ObjectMeta: metav1.ObjectMeta{ Name: svcName(cn), + }}, &corev1.Service{ObjectMeta: metav1.ObjectMeta{ + Name: metricSvcName(cn), }}} for _, obj := range objs { obj.SetNamespace(cn.Namespace) diff --git a/pkg/controllers/cnset/controller_test.go b/pkg/controllers/cnset/controller_test.go index 279442ae..3ddf3fdf 100644 --- a/pkg/controllers/cnset/controller_test.go +++ b/pkg/controllers/cnset/controller_test.go @@ -1,4 +1,4 @@ -// Copyright 2025 Matrix Origin +// Copyright 2025-2026 Matrix Origin // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -15,6 +15,8 @@ package cnset import ( + "context" + "strconv" "testing" kruisev1alpha1 "github.com/openkruise/kruise-api/apps/v1alpha1" @@ -37,6 +39,148 @@ import ( "sigs.k8s.io/controller-runtime/pkg/client" ) +func baseCNSetForMetricSvcTest() *v1alpha1.CNSet { + return &v1alpha1.CNSet{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "default", + Name: "test", + UID: "test-uid", + }, + Spec: v1alpha1.CNSetSpec{ + PodSet: v1alpha1.PodSet{ + MainContainer: v1alpha1.MainContainer{ + Image: "test:latest", + }, + Replicas: 1, + }, + }, + } +} + +func enableCNPromServiceDiscovery(cn *v1alpha1.CNSet) { + cn.Spec.ExportToPrometheus = pointer.Bool(true) + scheme := v1alpha1.PromDiscoverySchemeService + cn.Spec.PromDiscoveryScheme = &scheme +} + +// Test_syncMetricService is a regression test for issue #600: CNSet must reconcile a +// dedicated metric Service so ServiceMonitor can match on port name "metric". +func Test_syncMetricService(t *testing.T) { + s := newScheme() + labels := common.SubResourceLabels(baseCNSetForMetricSvcTest()) + + tests := []struct { + name string + cnset *v1alpha1.CNSet + client client.Client + setup func(cn *v1alpha1.CNSet) + expect func(g *WithT, cn *v1alpha1.CNSet, cli client.Client, err error) + }{ + { + name: "creates metric service with prom annotations when export enabled", + cnset: baseCNSetForMetricSvcTest(), + client: &fake.Client{ + Client: fake.KubeClientBuilder().WithScheme(s).Build(), + }, + setup: enableCNPromServiceDiscovery, + expect: func(g *WithT, cn *v1alpha1.CNSet, cli client.Client, err error) { + g.Expect(err).To(BeNil()) + + svc := &corev1.Service{} + g.Expect(cli.Get(context.Background(), client.ObjectKey{ + Namespace: cn.Namespace, + Name: metricSvcName(cn), + }, svc)).To(Succeed()) + + g.Expect(svc.Spec.Type).To(Equal(corev1.ServiceTypeClusterIP)) + g.Expect(svc.Spec.Ports).To(HaveLen(1)) + g.Expect(svc.Spec.Ports[0].Name).To(Equal("metric")) + g.Expect(svc.Spec.Ports[0].Port).To(Equal(int32(common.MetricsPort))) + g.Expect(svc.Labels).To(Equal(labels)) + g.Expect(svc.Spec.Selector).To(Equal(labels)) + g.Expect(svc.Annotations[common.PrometheusScrapeAnno]).To(Equal("true")) + g.Expect(svc.Annotations[common.PrometheusPortAnno]).To(Equal(strconv.Itoa(common.MetricsPort))) + }, + }, + { + name: "creates metric service without prom annotations when export disabled", + cnset: baseCNSetForMetricSvcTest(), + client: &fake.Client{ + Client: fake.KubeClientBuilder().WithScheme(s).Build(), + }, + expect: func(g *WithT, cn *v1alpha1.CNSet, cli client.Client, err error) { + g.Expect(err).To(BeNil()) + + svc := &corev1.Service{} + g.Expect(cli.Get(context.Background(), client.ObjectKey{ + Namespace: cn.Namespace, + Name: metricSvcName(cn), + }, svc)).To(Succeed()) + + g.Expect(svc.Spec.Ports).To(HaveLen(1)) + g.Expect(svc.Spec.Ports[0].Name).To(Equal("metric")) + g.Expect(svc.Annotations).NotTo(HaveKey(common.PrometheusScrapeAnno)) + g.Expect(svc.Annotations).NotTo(HaveKey(common.PrometheusPortAnno)) + }, + }, + { + name: "removes stale prom annotations when export disabled", + cnset: baseCNSetForMetricSvcTest(), + client: &fake.Client{ + Client: fake.KubeClientBuilder().WithScheme(s).WithObjects( + &corev1.Service{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "default", + Name: metricSvcName(baseCNSetForMetricSvcTest()), + Labels: labels, + Annotations: map[string]string{ + common.PrometheusScrapeAnno: "true", + common.PrometheusPortAnno: strconv.Itoa(common.MetricsPort), + }, + }, + Spec: corev1.ServiceSpec{ + Type: corev1.ServiceTypeClusterIP, + Selector: labels, + Ports: []corev1.ServicePort{{ + Name: "metric", + Port: int32(common.MetricsPort), + }}, + }, + }, + ).Build(), + }, + expect: func(g *WithT, cn *v1alpha1.CNSet, cli client.Client, err error) { + g.Expect(err).To(BeNil()) + + svc := &corev1.Service{} + g.Expect(cli.Get(context.Background(), client.ObjectKey{ + Namespace: cn.Namespace, + Name: metricSvcName(cn), + }, svc)).To(Succeed()) + g.Expect(svc.Annotations).NotTo(HaveKey(common.PrometheusScrapeAnno)) + g.Expect(svc.Annotations).NotTo(HaveKey(common.PrometheusPortAnno)) + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + g := NewGomegaWithT(t) + cn := tt.cnset.DeepCopy() + if tt.setup != nil { + tt.setup(cn) + } + + mockCtrl := gomock.NewController(t) + eventEmitter := fake.NewMockEventEmitter(mockCtrl) + ctx := fake.NewContext(cn, tt.client, eventEmitter) + + err := (&Actor{}).syncMetricService(ctx) + tt.expect(g, cn, tt.client, err) + }) + } +} + func TestCNSetActor_Observe(t *testing.T) { s := newScheme() tpl := &v1alpha1.CNSet{ diff --git a/pkg/controllers/cnset/resource.go b/pkg/controllers/cnset/resource.go index dfa2d8b3..de3a3035 100644 --- a/pkg/controllers/cnset/resource.go +++ b/pkg/controllers/cnset/resource.go @@ -157,6 +157,7 @@ func syncService(cn *v1alpha1.CNSet, svc *corev1.Service) { svc.Annotations[common.PrometheusPortAnno] = strconv.Itoa(common.MetricsPort) } else { delete(svc.Annotations, common.PrometheusScrapeAnno) + delete(svc.Annotations, common.PrometheusPortAnno) } } diff --git a/pkg/controllers/cnset/utils.go b/pkg/controllers/cnset/utils.go index f6b8daef..0a2a5309 100644 --- a/pkg/controllers/cnset/utils.go +++ b/pkg/controllers/cnset/utils.go @@ -42,6 +42,10 @@ func svcName(cn *v1alpha1.CNSet) string { return resourceName(cn) } +func metricSvcName(cn *v1alpha1.CNSet) string { + return resourceName(cn) + "-metric" +} + func setName(cn *v1alpha1.CNSet) string { return resourceName(cn) }