feat: Add replicas field to Function CRD for configurable deployment scaling (#202)

### Motivation

Currently, Function deployments are hardcoded to use 1 replica, which
limits the ability to scale functions based on workload requirements.
This change adds a configurable `replicas` field to the Function CRD,
allowing users to specify the desired number of replicas for their
function deployments. This enables horizontal scaling of functions to
handle increased load and improve availability.

### Modifications

- **Function CRD Enhancement**: Added `replicas` field to `FunctionSpec`
with validation and default value of 1
- **Controller Logic**: Updated `FunctionReconciler` to use the
specified replicas when creating deployments
- **CRD Generation**: Updated generated deepcopy functions and CRD YAML
files to include the new field
- **Testing**: Added comprehensive test cases to verify replica
configuration works correctly, including:
  - Custom replica specification
  - Replica updates
  - Default replica behavior when not specified
- **Gitignore Updates**: Added macOS-specific ignore patterns
(`.DS_Store` and `._*` files) to prevent accidental commits

The implementation maintains backward compatibility by defaulting to 1
replica when the field is not specified, ensuring existing function
deployments continue to work without modification.

Author: evanwave

.gitignore

@@ -110,4 +110,7 @@ bin/
 
 benchmark/*.pprof
 
-operator/vendor/
+operator/vendor/
+
+._*
+**/.DS_Store

operator/api/v1alpha1/function_types.go

@@ -37,6 +37,10 @@ type FunctionSpec struct {
 	// Module name
 	// +kubebuilder:validation:Required
 	Module string `json:"module"`
+	// Number of replicas for the function deployment
+	// +kubebuilder:validation:Optional
+	// +kubebuilder:default=1
+	Replicas *int32 `json:"replicas,omitempty"`
 	// +kubebuilder:validation:Optional
 	SubscriptionName string `json:"subscriptionName,omitempty"`
 	// List of sources

operator/api/v1alpha1/zz_generated.deepcopy.go

@@ -56,6 +56,11 @@ spec:
               package:
                 description: Package name
                 type: string
+              replicas:
+                default: 1
+                description: Number of replicas for the function deployment
+                format: int32
+                type: integer
               requestSource:
                 description: Request source
                 properties:

operator/config/crd/bases/fs.functionstream.github.io_functions.yaml

@@ -62,6 +62,11 @@ spec:
               package:
                 description: Package name
                 type: string
+              replicas:
+                default: 1
+                description: Number of replicas for the function deployment
+                format: int32
+                type: integer
               requestSource:
                 description: Request source
                 properties:

operator/deploy/chart/templates/crd/fs.functionstream.github.io_functions.yaml

@@ -118,6 +118,9 @@ func (r *FunctionReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 	// 5. Build Deployment
 	deployName := fmt.Sprintf("function-%s", fn.Name)
 	var replicas int32 = 1
+	if fn.Spec.Replicas != nil {
+		replicas = *fn.Spec.Replicas
+	}
 	labels := map[string]string{
 		"function": fn.Name,
 	}

operator/internal/controller/function_controller.go

@@ -581,6 +581,112 @@ var _ = Describe("Function Controller", func() {
 			Expect(deploy2.Spec.Template.Spec.Containers[0].Image).To(Equal("nginx:latest"))
 		})
 
+		It("should use the specified replicas from FunctionSpec", func() {
+			By("creating a Function with custom replicas")
+			controllerReconciler := &FunctionReconciler{
+				Client: k8sClient,
+				Scheme: k8sClient.Scheme(),
+				Config: Config{
+					PulsarServiceURL: "pulsar://test-broker:6650",
+					PulsarAuthPlugin: "org.apache.pulsar.client.impl.auth.AuthenticationToken",
+					PulsarAuthParams: "token:my-token",
+				},
+			}
+
+			// Create a Package resource first
+			pkg := &fsv1alpha1.Package{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-pkg-replicas",
+					Namespace: "default",
+				},
+				Spec: fsv1alpha1.PackageSpec{
+					DisplayName: "Test Package Replicas",
+					Description: "desc",
+					FunctionType: fsv1alpha1.FunctionType{
+						Cloud: &fsv1alpha1.CloudType{Image: "busybox:latest"},
+					},
+					Modules: map[string]fsv1alpha1.Module{},
+				},
+			}
+			Expect(k8sClient.Create(ctx, pkg)).To(Succeed())
+
+			// Create a Function with custom replicas
+			customReplicas := int32(3)
+			fn := &fsv1alpha1.Function{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-fn-replicas",
+					Namespace: "default",
+				},
+				Spec: fsv1alpha1.FunctionSpec{
+					Package:          "test-pkg-replicas",
+					Module:           "mod",
+					Replicas:         &customReplicas,
+					SubscriptionName: "sub",
+					Sink:             &fsv1alpha1.SinkSpec{Pulsar: &fsv1alpha1.PulsarSinkSpec{Topic: "out"}},
+					RequestSource:    &fsv1alpha1.SourceSpec{Pulsar: &fsv1alpha1.PulsarSourceSpec{Topic: "in"}},
+				},
+			}
+			Expect(k8sClient.Create(ctx, fn)).To(Succeed())
+
+			// Reconcile the Function
+			_, err := controllerReconciler.Reconcile(ctx, reconcile.Request{
+				NamespacedName: types.NamespacedName{Name: fn.Name, Namespace: fn.Namespace},
+			})
+			Expect(err).NotTo(HaveOccurred())
+
+			// Check that the Deployment has the correct number of replicas
+			deployName := "function-" + fn.Name
+			deploy := &appsv1.Deployment{}
+			Expect(k8sClient.Get(ctx, types.NamespacedName{Name: deployName, Namespace: fn.Namespace}, deploy)).To(Succeed())
+			Expect(deploy.Spec.Replicas).NotTo(BeNil())
+			Expect(*deploy.Spec.Replicas).To(Equal(int32(3)))
+
+			// Test updating replicas
+			newReplicas := int32(5)
+			patch := client.MergeFrom(fn.DeepCopy())
+			fn.Spec.Replicas = &newReplicas
+			Expect(k8sClient.Patch(ctx, fn, patch)).To(Succeed())
+
+			// Reconcile again
+			_, err = controllerReconciler.Reconcile(ctx, reconcile.Request{
+				NamespacedName: types.NamespacedName{Name: fn.Name, Namespace: fn.Namespace},
+			})
+			Expect(err).NotTo(HaveOccurred())
+
+			// Verify the deployment was updated with new replicas
+			Expect(k8sClient.Get(ctx, types.NamespacedName{Name: deployName, Namespace: fn.Namespace}, deploy)).To(Succeed())
+			Expect(*deploy.Spec.Replicas).To(Equal(int32(5)))
+
+			// Test default replicas when not specified
+			fnDefault := &fsv1alpha1.Function{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-fn-default-replicas",
+					Namespace: "default",
+				},
+				Spec: fsv1alpha1.FunctionSpec{
+					Package:          "test-pkg-replicas",
+					Module:           "mod",
+					SubscriptionName: "sub-default",
+					Sink:             &fsv1alpha1.SinkSpec{Pulsar: &fsv1alpha1.PulsarSinkSpec{Topic: "out-default"}},
+					RequestSource:    &fsv1alpha1.SourceSpec{Pulsar: &fsv1alpha1.PulsarSourceSpec{Topic: "in-default"}},
+				},
+			}
+			Expect(k8sClient.Create(ctx, fnDefault)).To(Succeed())
+
+			// Reconcile the Function with default replicas
+			_, err = controllerReconciler.Reconcile(ctx, reconcile.Request{
+				NamespacedName: types.NamespacedName{Name: fnDefault.Name, Namespace: fnDefault.Namespace},
+			})
+			Expect(err).NotTo(HaveOccurred())
+
+			// Check that the Deployment has default replicas (1)
+			deployDefaultName := "function-" + fnDefault.Name
+			deployDefault := &appsv1.Deployment{}
+			Expect(k8sClient.Get(ctx, types.NamespacedName{Name: deployDefaultName, Namespace: fnDefault.Namespace}, deployDefault)).To(Succeed())
+			Expect(deployDefault.Spec.Replicas).NotTo(BeNil())
+			Expect(*deployDefault.Spec.Replicas).To(Equal(int32(1)))
+		})
+
 		It("should handle Package updates in different namespaces", func() {
 			By("creating Functions and Packages in different namespaces")
 			controllerReconciler := &FunctionReconciler{

operator/internal/controller/function_controller_test.go

@@ -83,6 +83,11 @@ spec:
               package:
                 description: Package name
                 type: string
+              replicas:
+                default: 1
+                description: Number of replicas for the function deployment
+                format: int32
+                type: integer
               requestSource:
                 description: Request source
                 properties: