fix: fix service type of standalone service, optimize for telemetry (#…

…576) * fix: fix service type of standalone service, optimize for telemetry Signed-off-by: arkbriar <[email protected]> * test: add test cases for standalone services Signed-off-by: arkbriar <[email protected]> --------- Signed-off-by: arkbriar <[email protected]>
risingwavelabs · Jan 10, 2024 · ee84c58 · ee84c58
1 parent a1bef20
commit ee84c58
Show file tree

Hide file tree

Showing 4 changed files with 46 additions and 15 deletions.
diff --git a/pkg/factory/risingwave_object_factory.go b/pkg/factory/risingwave_object_factory.go
@@ -1716,7 +1716,7 @@ func (f *RisingWaveObjectFactory) NewStandaloneAdvancedStatefulSet() *kruiseapps
 
 // NewStandaloneService creates a Service for standalone component.
 func (f *RisingWaveObjectFactory) NewStandaloneService() *corev1.Service {
-	standaloneSvc := f.newService(consts.ComponentStandalone, f.risingwave.Spec.FrontendServiceType, []corev1.ServicePort{
+	standaloneSvc := f.newService(consts.ComponentStandalone, corev1.ServiceTypeClusterIP, []corev1.ServicePort{
 		{
 			Name:       consts.PortService,
 			Protocol:   corev1.ProtocolTCP,
@@ -1771,20 +1771,27 @@ func (f *RisingWaveObjectFactory) NewMetaService() *corev1.Service {
 
 // NewFrontendService creates a new Service for the frontend.
 func (f *RisingWaveObjectFactory) NewFrontendService() *corev1.Service {
-	frontendSvc := f.newService(consts.ComponentFrontend, f.risingwave.Spec.FrontendServiceType, []corev1.ServicePort{
+	svcPorts := []corev1.ServicePort{
 		{
 			Name:       consts.PortService,
 			Protocol:   corev1.ProtocolTCP,
 			Port:       consts.FrontendServicePort,
 			TargetPort: intstr.FromString(consts.PortService),
 		},
-		{
+	}
+
+	// Disable metrics port when RisingWave is in standalone mode. This makes it
+	// easier to write a single telemetry configuration to cover all running RisingWaves.
+	if !ptr.Deref(f.risingwave.Spec.EnableStandaloneMode, false) {
+		svcPorts = append(svcPorts, corev1.ServicePort{
 			Name:       consts.PortMetrics,
 			Protocol:   corev1.ProtocolTCP,
 			Port:       consts.FrontendMetricsPort,
 			TargetPort: intstr.FromString(consts.PortMetrics),
-		},
-	})
+		})
+	}
+
+	frontendSvc := f.newService(consts.ComponentFrontend, f.risingwave.Spec.FrontendServiceType, svcPorts)
 
 	// Hijack selector if it's standalone mode.
 	if object.NewRisingWaveReader(f.risingwave).IsStandaloneModeEnabled() {
@@ -2036,11 +2043,6 @@ func (f *RisingWaveObjectFactory) NewServiceMonitor() *prometheusv1.ServiceMonit
 						Operator: metav1.LabelSelectorOpIn,
 						Values:   []string{f.risingwave.Name},
 					},
-					{
-						Key:      consts.LabelRisingWaveComponent,
-						Operator: metav1.LabelSelectorOpNotIn,
-						Values:   []string{consts.ComponentStandalone},
-					},
 				},
 			},
 		},

diff --git a/pkg/factory/risingwave_object_factory_predicate_test.go b/pkg/factory/risingwave_object_factory_predicate_test.go
@@ -934,7 +934,8 @@ func servicesPredicates() []predicate[*corev1.Service, servicesTestCase] {
 		{
 			Name: "selector-equals",
 			Fn: func(obj *corev1.Service, testcase servicesTestCase) bool {
-				return hasServiceSelector(obj, podSelector(testcase.risingwave, testcase.component, nil))
+				selectorComponent := lo.If(testcase.selectorComponent == "", testcase.component).Else(testcase.selectorComponent)
+				return hasServiceSelector(obj, podSelector(testcase.risingwave, selectorComponent, nil))
 			},
 		},
 	}

diff --git a/pkg/factory/risingwave_object_factory_test.go b/pkg/factory/risingwave_object_factory_test.go
@@ -34,6 +34,7 @@ func Test_RisingWaveObjectFactory_Services(t *testing.T) {
 
 	for name, tc := range servicesTestCases() {
 		tc.risingwave = newTestRisingwave(func(r *risingwavev1alpha1.RisingWave) {
+			r.Spec.EnableStandaloneMode = ptr.To(tc.enableStandaloneMode)
 			r.Spec.FrontendServiceType = tc.globalServiceType
 		})
 
@@ -51,6 +52,8 @@ func Test_RisingWaveObjectFactory_Services(t *testing.T) {
 			svc = factory.NewCompactorService()
 		case consts.ComponentConnector:
 			svc = factory.NewConnectorService()
+		case consts.ComponentStandalone:
+			svc = factory.NewStandaloneService()
 		default:
 			t.Fatal("bad test")
 		}
@@ -83,6 +86,8 @@ func Test_RisingWaveObjectFactory_ServicesMeta(t *testing.T) {
 			svc = factory.NewCompactorService()
 		case consts.ComponentConnector:
 			svc = factory.NewConnectorService()
+		case consts.ComponentStandalone:
+			svc = factory.NewStandaloneService()
 		default:
 			t.Fatal("bad test")
 		}

diff --git a/pkg/factory/risingwave_object_factory_testcases_test.go b/pkg/factory/risingwave_object_factory_testcases_test.go
@@ -2694,10 +2694,12 @@ func computeAdvancedSTSTestCases() map[string]computeAdvancedSTSTestCase {
 
 type servicesTestCase struct {
 	baseTestCase
-	component         string
-	ports             map[string]int32
-	globalServiceType corev1.ServiceType
-	expectServiceType corev1.ServiceType
+	component            string
+	selectorComponent    string // Empty means equals to component.
+	ports                map[string]int32
+	enableStandaloneMode bool
+	globalServiceType    corev1.ServiceType
+	expectServiceType    corev1.ServiceType
 }
 
 func servicesTestCases() map[string]servicesTestCase {
@@ -2773,6 +2775,27 @@ func servicesTestCases() map[string]servicesTestCase {
 			globalServiceType: corev1.ServiceTypeNodePort,
 			expectServiceType: corev1.ServiceTypeClusterIP,
 		},
+		"standalone-ports": {
+			component:            consts.ComponentStandalone,
+			enableStandaloneMode: true,
+			globalServiceType:    corev1.ServiceTypeNodePort,
+			expectServiceType:    corev1.ServiceTypeClusterIP,
+			ports: map[string]int32{
+				consts.PortService:   consts.FrontendServicePort,
+				consts.PortMetrics:   consts.MetaMetricsPort,
+				consts.PortDashboard: consts.MetaDashboardPort,
+			},
+		},
+		"standalone-frontend-ports": {
+			component:            consts.ComponentFrontend,
+			selectorComponent:    consts.ComponentStandalone,
+			enableStandaloneMode: true,
+			globalServiceType:    corev1.ServiceTypeNodePort,
+			expectServiceType:    corev1.ServiceTypeNodePort,
+			ports: map[string]int32{
+				consts.PortService: consts.FrontendServicePort,
+			},
+		},
 	}
 
 }