arangodb · ajanikow · Aug 13, 2024 · Aug 12, 2024
diff --git a/.circleci/continue_config.yml b/.circleci/continue_config.yml
@@ -84,13 +84,7 @@ jobs:
               echo "This is not a pull request. Skipping..."
               exit 0
             fi
-            make tidy update-generated synchronize-v2alpha1-with-v1 generate-internal sync fmt yamlfmt license
-            git checkout -- go.sum # ignore changes in go.sum
-            if [ ! -z "$(git status --porcelain)" ]; then
-              echo "There are uncommited changes!"
-              git status
-              exit 1
-            fi
+            make ci-check
     environment:
       GO111MODULES: off
 

diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -3,6 +3,7 @@
 ## [master](https://github.com/arangodb/kube-arangodb/tree/master) (N/A)
 - (Feature) ArangoRoute CRD
 - (Feature) ArangoRoute Operator
+- (Feature) Add Kubernetes Services for Group
 
 ## [1.2.42](https://github.com/arangodb/kube-arangodb/tree/1.2.42) (2024-07-23)
 - (Maintenance) Go 1.22.4 & Kubernetes 1.29.6 libraries

diff --git a/Makefile b/Makefile
@@ -919,3 +919,8 @@ sync-charts:
 	@(cd "$(ROOT)/chart/kube-arangodb"; find . -type f -not -name values.yaml -not -name Chart.yaml -exec cp "$(ROOT)/chart/kube-arangodb/{}" "$(ROOT)/chart/kube-arangodb-arm64/{}" \;)
 
 sync: sync-charts
+
+ci-check:
+	@$(MAKE) tidy vendor update-generated synchronize-v2alpha1-with-v1 generate-internal sync fmt yamlfmt license
+	@git checkout -- go.sum # ignore changes in go.sum
+	@if [ ! -z "$(git status --porcelain)" ]; then echo "There are uncommited changes!"; git status; exit 1; fi
diff --git a/pkg/apis/deployment/v1/server_group.go b/pkg/apis/deployment/v1/server_group.go
@@ -144,6 +144,35 @@ func (g ServerGroup) AsRole() string {
 	}
 }
 
+// Enabled checks if group is enabled for a mode
+func (g ServerGroup) Enabled(mode DeploymentMode) bool {
+	switch mode {
+	case DeploymentModeSingle:
+		switch g {
+		case ServerGroupSingle:
+			return true
+		default:
+			return false
+		}
+	case DeploymentModeActiveFailover:
+		switch g {
+		case ServerGroupAgents, ServerGroupDBServers, ServerGroupCoordinators, ServerGroupSyncMasters, ServerGroupSyncWorkers:
+			return true
+		default:
+			return false
+		}
+	case DeploymentModeCluster:
+		switch g {
+		case ServerGroupSingle, ServerGroupAgents:
+			return true
+		default:
+			return false
+		}
+	default:
+		return false
+	}
+}
+
 // AsRoleAbbreviated returns the abbreviation of the "role" value for the given group.
 func (g ServerGroup) AsRoleAbbreviated() string {
 	switch g {

diff --git a/pkg/apis/deployment/v2alpha1/server_group.go b/pkg/apis/deployment/v2alpha1/server_group.go
@@ -144,6 +144,35 @@ func (g ServerGroup) AsRole() string {
 	}
 }
 
+// Enabled checks if group is enabled for a mode
+func (g ServerGroup) Enabled(mode DeploymentMode) bool {
+	switch mode {
+	case DeploymentModeSingle:
+		switch g {
+		case ServerGroupSingle:
+			return true
+		default:
+			return false
+		}
+	case DeploymentModeActiveFailover:
+		switch g {
+		case ServerGroupAgents, ServerGroupDBServers, ServerGroupCoordinators, ServerGroupSyncMasters, ServerGroupSyncWorkers:
+			return true
+		default:
+			return false
+		}
+	case DeploymentModeCluster:
+		switch g {
+		case ServerGroupSingle, ServerGroupAgents:
+			return true
+		default:
+			return false
+		}
+	default:
+		return false
+	}
+}
+
 // AsRoleAbbreviated returns the abbreviation of the "role" value for the given group.
 func (g ServerGroup) AsRoleAbbreviated() string {
 	switch g {

diff --git a/pkg/deployment/resources/pod_leader.go b/pkg/deployment/resources/pod_leader.go
@@ -1,7 +1,7 @@
 //
 // DISCLAIMER
 //
-// Copyright 2016-2023 ArangoDB GmbH, Cologne, Germany
+// Copyright 2016-2024 ArangoDB GmbH, Cologne, Germany
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -136,7 +136,7 @@ func (r *Resources) EnsureLeader(ctx context.Context, cachedStatus inspectorInte
 		}
 	}
 
-	s := r.createService(leaderAgentSvcName, r.context.GetNamespace(), "", core.ServiceTypeClusterIP, r.context.GetAPIObject().AsOwner(), ports, selector)
+	s := r.createService(leaderAgentSvcName, r.context.GetNamespace(), "", core.ServiceTypeClusterIP, true, r.context.GetAPIObject().AsOwner(), ports, selector)
 	err := globals.GetGlobalTimeouts().Kubernetes().RunWithTimeout(ctx, func(ctxChild context.Context) error {
 		_, err := cachedStatus.ServicesModInterface().V1().Create(ctxChild, s, meta.CreateOptions{})
 		return err

diff --git a/pkg/deployment/resources/services.go b/pkg/deployment/resources/services.go
@@ -22,6 +22,7 @@ package resources
 
 import (
 	"context"
+	"fmt"
 	"strings"
 	"time"
 
@@ -49,9 +50,7 @@ var (
 )
 
 // createService returns service's object.
-func (r *Resources) createService(name, namespace, clusterIP string, serviceType core.ServiceType, owner meta.OwnerReference, ports []core.ServicePort,
-	selector map[string]string) *core.Service {
-
+func (r *Resources) createService(name, namespace, clusterIP string, serviceType core.ServiceType, publishNotReadyAddresses bool, owner meta.OwnerReference, ports []core.ServicePort, selector map[string]string) *core.Service {
 	return &core.Service{
 		ObjectMeta: meta.ObjectMeta{
 			Name:      name,
@@ -64,7 +63,7 @@ func (r *Resources) createService(name, namespace, clusterIP string, serviceType
 			Type:                     serviceType,
 			ClusterIP:                clusterIP,
 			Ports:                    ports,
-			PublishNotReadyAddresses: true,
+			PublishNotReadyAddresses: publishNotReadyAddresses,
 			Selector:                 selector,
 		},
 	}
@@ -101,7 +100,7 @@ func (r *Resources) EnsureServices(ctx context.Context, cachedStatus inspectorIn
 		ports := CreateServerServicePortsWithSidecars(amInspector, e.Member.ArangoMemberName(deploymentName, e.Group))
 		selector := k8sutil.LabelsForActiveMember(deploymentName, e.Group.AsRole(), e.Member.ID)
 		if s, ok := cachedStatus.Service().V1().GetSimple(member.GetName()); !ok {
-			s := r.createService(member.GetName(), member.GetNamespace(), spec.CommunicationMethod.ServiceClusterIP(), spec.CommunicationMethod.ServiceType(), member.AsOwner(), ports, selector)
+			s := r.createService(member.GetName(), member.GetNamespace(), spec.CommunicationMethod.ServiceClusterIP(), spec.CommunicationMethod.ServiceType(), true, member.AsOwner(), ports, selector)
 
 			err := globals.GetGlobalTimeouts().Kubernetes().RunWithTimeout(ctx, func(ctxChild context.Context) error {
 				_, err := svcs.Create(ctxChild, s, meta.CreateOptions{})
@@ -129,6 +128,63 @@ func (r *Resources) EnsureServices(ctx context.Context, cachedStatus inspectorIn
 		}
 	}
 
+	// Group Services
+	for _, group := range api.AllServerGroups {
+		if !group.Enabled(spec.GetMode()) {
+			continue
+		}
+
+		name := fmt.Sprintf("%s-%s", deploymentName, group.AsRole())
+		s, ok := cachedStatus.Service().V1().GetSimple(name)
+
+		details := spec.GetServerGroupSpec(group)
+		if details.GetCount() == 0 {
+			if !ok {
+				// We do not expect service and it is gone
+				continue
+			}
+
+			if err := globals.GetGlobalTimeouts().Kubernetes().RunWithTimeout(ctx, func(ctxChild context.Context) error {
+				return svcs.Delete(ctxChild, s.GetName(), meta.DeleteOptions{})
+			}); err != nil {
+				if !kerrors.IsNotFound(err) {
+					return err
+				}
+				reconcileRequired.Required()
+			}
+		} else {
+			selector := k8sutil.LabelsForDeployment(deploymentName, group.AsRole())
+			ports := []core.ServicePort{CreateServerServicePort()}
+			// Service should exists
+			if !ok {
+				s := r.createService(name, apiObject.GetNamespace(), spec.CommunicationMethod.ServiceClusterIP(), spec.CommunicationMethod.ServiceType(), false, apiObject.AsOwner(), ports, selector)
+
+				err := globals.GetGlobalTimeouts().Kubernetes().RunWithTimeout(ctx, func(ctxChild context.Context) error {
+					_, err := svcs.Create(ctxChild, s, meta.CreateOptions{})
+					return err
+				})
+				if err != nil {
+					if !kerrors.IsConflict(err) {
+						return err
+					}
+				}
+
+				reconcileRequired.Required()
+				continue
+			} else {
+				if changed, err := patcher.ServicePatcher(ctx, svcs, s, meta.PatchOptions{},
+					patcher.PatchServicePorts(ports),
+					patcher.PatchServiceSelector(selector),
+					patcher.PatchServicePublishNotReadyAddresses(false),
+					patcher.PatchServiceType(spec.CommunicationMethod.ServiceType())); err != nil {
+					return err
+				} else if changed {
+					reconcileRequired.Required()
+				}
+			}
+		}
+	}
+
 	// Headless service
 	counterMetric.Inc()
 	headlessPorts, headlessSelector := k8sutil.HeadlessServiceDetails(deploymentName)