Add scaling support

api/v1alpha1/asset_types.go

@@ -33,10 +33,15 @@ type AssetStatus struct {
 	// INSERT ADDITIONAL STATUS FIELD - define observed state of cluster
 	// Important: Run "make" to regenerate code after modifying this file
 	Conditions []metav1.Condition `json:"conditions,omitempty"`
+	// Replicas is the number of actual replicas of the asset deployment
+	Replicas int32 `json:"replicas,omitempty"`
+	// Selector is the label selector for pods corresponding to this asset deployment
+	Selector string `json:"selector,omitempty"`
 }
 
 //+kubebuilder:object:root=true
 //+kubebuilder:subresource:status
+//+kubebuilder:subresource:scale:specpath=.spec.deploymentOverrides.replicas,statuspath=.status.replicas,selectorpath=.status.selector
 
 // Asset is the Schema for the assets API
 type Asset struct {

api/v1alpha1/propagation_types.go

@@ -34,10 +34,19 @@ type PropagationSpec struct {
 // PropagationStatus defines the observed state of Propagation
 type PropagationStatus struct {
 	Conditions []metav1.Condition `json:"conditions,omitempty"`
+	// Replicas is the number of actual replicas of the propagation deployment
+	Replicas int32 `json:"replicas,omitempty"`
+	// Selector is the label selector for pods corresponding to this propagation deployment
+	Selector string `json:"selector,omitempty"`
 }
 
 //+kubebuilder:object:root=true
 //+kubebuilder:subresource:status
+//+kubebuilder:subresource:scale:specpath=.spec.deploymentOverrides.replicas,statuspath=.status.replicas,selectorpath=.status.selector
+//+kubebuilder:printcolumn:name="Desired",type=integer,JSONPath=`.spec.deploymentOverrides.replicas`,description="Desired number of replicas"
+//+kubebuilder:printcolumn:name="Current",type=integer,JSONPath=`.status.replicas`,description="Current number of replicas"
+//+kubebuilder:printcolumn:name="Ready",type=string,JSONPath=`.status.conditions[?(@.type=="Reconciled")].status`,description="Ready status"
+//+kubebuilder:printcolumn:name="Age",type=date,JSONPath=`.metadata.creationTimestamp`
 
 // Propagation is the Schema for the propagations API
 type Propagation struct {

api/v1alpha1/subtreevalidator_types.go

@@ -31,10 +31,15 @@ type SubtreeValidatorStatus struct {
 	// INSERT ADDITIONAL STATUS FIELD - define observed state of cluster
 	// Important: Run "make" to regenerate code after modifying this file
 	Conditions []metav1.Condition `json:"conditions,omitempty"`
+	// Replicas is the number of actual replicas of the subtreevalidator deployment
+	Replicas int32 `json:"replicas,omitempty"`
+	// Selector is the label selector for pods corresponding to this subtreevalidator deployment
+	Selector string `json:"selector,omitempty"`
 }
 
 //+kubebuilder:object:root=true
 //+kubebuilder:subresource:status
+//+kubebuilder:subresource:scale:specpath=.spec.deploymentOverrides.replicas,statuspath=.status.replicas,selectorpath=.status.selector
 
 // SubtreeValidator is the Schema for the subtreevalidators API
 type SubtreeValidator struct {

cmd/main.go

@@ -25,9 +25,12 @@ import (
 	"sort"
 	"strings"
 
+	teranodev1alpha1 "github.com/bsv-blockchain/teranode-operator/api/v1alpha1"
+	appsv1 "k8s.io/api/apps/v1"
+	scalev1 "k8s.io/api/autoscaling/v1"
+	corev1 "k8s.io/api/core/v1"
+	networkingv1 "k8s.io/api/networking/v1"
 	"k8s.io/apimachinery/pkg/runtime"
-	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
-	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 
 	// Import all Kubernetes client auth plugins (e.g. Azure, GCP, OIDC, etc.)
 	// to ensure that exec-entrypoint and run can make use of them.
@@ -39,7 +42,6 @@ import (
 	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
 	"sigs.k8s.io/controller-runtime/pkg/webhook"
 
-	teranodev1alpha1 "github.com/bsv-blockchain/teranode-operator/api/v1alpha1"
 	"github.com/bsv-blockchain/teranode-operator/internal/controller"
 )
 
@@ -55,13 +57,6 @@ const (
 	CreateControllerError = "unable to create controller"
 )
 
-func init() {
-	utilruntime.Must(clientgoscheme.AddToScheme(scheme))
-
-	utilruntime.Must(teranodev1alpha1.AddToScheme(scheme))
-	//+kubebuilder:scaffold:scheme
-}
-
 //nolint:gocognit,gocyclo // Main function complexity is acceptable for initialization
 func main() {
 	var metricsAddr string
@@ -70,7 +65,7 @@ func main() {
 	var secureMetrics bool
 	var enableHTTP2 bool
 	flag.StringVar(&metricsAddr, "metrics-bind-address", ":8080", "The address the metric endpoint binds to.")
-	flag.StringVar(&probeAddr, "health-probe-bind-address", ":8081", "The address the probe endpoint binds to.")
+	flag.StringVar(&probeAddr, "health-probe-bind-address", ":8181", "The address the probe endpoint binds to.")
 	flag.BoolVar(&enableLeaderElection, "leader-elect", false,
 		"Enable leader election for controller manager. "+
 			"Enabling this will ensure there is only one active controller manager.")
@@ -84,6 +79,33 @@ func main() {
 	opts.BindFlags(flag.CommandLine)
 	flag.Parse()
 
+	if err := scalev1.AddToScheme(scheme); err != nil {
+		setupLog.Error(err, "unable to add autoscaling/v1 scheme")
+		os.Exit(1)
+	}
+
+	if err := teranodev1alpha1.AddToScheme(scheme); err != nil {
+		setupLog.Error(err, "unable to add v1alpha1 scheme")
+		os.Exit(1)
+	}
+
+	if err := appsv1.AddToScheme(scheme); err != nil {
+		setupLog.Error(err, "unable to add apps/v1 scheme")
+		os.Exit(1)
+	}
+
+	if err := corev1.AddToScheme(scheme); err != nil {
+		setupLog.Error(err, "unable to add core/v1 scheme")
+		os.Exit(1)
+	}
+
+	if err := networkingv1.AddToScheme(scheme); err != nil {
+		setupLog.Error(err, "unable to add networking/v1 scheme")
+		os.Exit(1)
+	}
+
+	//+kubebuilder:scaffold:scheme
+
 	ctrl.SetLogger(zap.New(zap.UseFlagOptions(&opts)))
 
 	// if the enable-http2 flag is false (the default), http/2 should be disabled

config/crd/bases/teranode.bsvblockchain.org_assets.yaml

@@ -3820,9 +3820,22 @@ spec:
                   - type
                   type: object
                 type: array
+              replicas:
+                description: Replicas is the number of actual replicas of the asset
+                  deployment
+                format: int32
+                type: integer
+              selector:
+                description: Selector is the label selector for pods corresponding
+                  to this asset deployment
+                type: string
             type: object
         type: object
     served: true
     storage: true
     subresources:
+      scale:
+        labelSelectorPath: .status.selector
+        specReplicasPath: .spec.deploymentOverrides.replicas
+        statusReplicasPath: .status.replicas
       status: {}

config/crd/bases/teranode.bsvblockchain.org_propagations.yaml

@@ -14,7 +14,23 @@ spec:
     singular: propagation
   scope: Namespaced
   versions:
-  - name: v1alpha1
+  - additionalPrinterColumns:
+    - description: Desired number of replicas
+      jsonPath: .spec.deploymentOverrides.replicas
+      name: Desired
+      type: integer
+    - description: Current number of replicas
+      jsonPath: .status.replicas
+      name: Current
+      type: integer
+    - description: Ready status
+      jsonPath: .status.conditions[?(@.type=="Reconciled")].status
+      name: Ready
+      type: string
+    - jsonPath: .metadata.creationTimestamp
+      name: Age
+      type: date
+    name: v1alpha1
     schema:
       openAPIV3Schema:
         description: Propagation is the Schema for the propagations API
@@ -3849,9 +3865,22 @@ spec:
                   - type
                   type: object
                 type: array
+              replicas:
+                description: Replicas is the number of actual replicas of the propagation
+                  deployment
+                format: int32
+                type: integer
+              selector:
+                description: Selector is the label selector for pods corresponding
+                  to this propagation deployment
+                type: string
             type: object
         type: object
     served: true
     storage: true
     subresources:
+      scale:
+        labelSelectorPath: .status.selector
+        specReplicasPath: .spec.deploymentOverrides.replicas
+        statusReplicasPath: .status.replicas
       status: {}

config/crd/bases/teranode.bsvblockchain.org_subtreevalidators.yaml

@@ -3782,9 +3782,22 @@ spec:
                   - type
                   type: object
                 type: array
+              replicas:
+                description: Replicas is the number of actual replicas of the subtreevalidator
+                  deployment
+                format: int32
+                type: integer
+              selector:
+                description: Selector is the label selector for pods corresponding
+                  to this subtreevalidator deployment
+                type: string
             type: object
         type: object
     served: true
     storage: true
     subresources:
+      scale:
+        labelSelectorPath: .status.selector
+        specReplicasPath: .spec.deploymentOverrides.replicas
+        statusReplicasPath: .status.replicas
       status: {}

config/rbac/teranode_service_runner_service_account.yaml

@@ -2,4 +2,4 @@ apiVersion: v1
 kind: ServiceAccount
 metadata:
   name: service-runner
-  namespace: system
+  namespace: system

config/samples/teranode_v1alpha1_propagation.yaml

@@ -9,11 +9,20 @@ metadata:
     app.kubernetes.io/created-by: teranode-operator
   name: propagation-sample
 spec:
-  # TODO(user): Add fields here
   resources:
     requests:
       cpu: 1
       memory: 1Gi
     limits:
       memory: 2Gi
   imagePullPolicy: Always
+  replicas: 1
+  deploymentOverrides:
+    replicas: 1
+    resources:
+      requests:
+        cpu: 1
+        memory: 1Gi
+      limits:
+        memory: 2Gi
+    imagePullPolicy: Always

internal/controller/asset_controller.go

@@ -26,6 +26,8 @@
 	networkingv1 "k8s.io/api/networking/v1"
 	apimeta "k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
+
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	ctrl "sigs.k8s.io/controller-runtime"
@@ -77,6 +79,17 @@
 		r.ReconcileHTTPSIngress,
 	)
 
+	// Update scale status (replicas and selector) from deployment
+	deployment := &appsv1.Deployment{}
+	if err := r.Get(ctx, types.NamespacedName{
+		Name:      AssetDeploymentName,
+		Namespace: asset.Namespace,
+	}, deployment); err == nil {
+		replicas, selector := utils.GetScaleStatusFromDeployment(deployment)
+		asset.Status.Replicas = replicas
+		asset.Status.Selector = selector
+	}
+
 	if err != nil {
 		apimeta.SetStatusCondition(&asset.Status.Conditions,
 			metav1.Condition{
@@ -103,6 +116,14 @@
 	}
 
 	err = r.Client.Status().Update(ctx, &asset)
+
+	if asset.Spec.DeploymentOverrides != nil && asset.Spec.DeploymentOverrides.Replicas != nil {
+		if asset.Status.Replicas != *asset.Spec.DeploymentOverrides.Replicas {
+			r.Log.Info("requeuing to monitor replica status", "status", asset.Status.Replicas, "spec", asset.Spec.DeploymentOverrides.Replicas)
+			return ctrl.Result{RequeueAfter: time.Second}, nil
+		}
+	}
+
 	return ctrl.Result{Requeue: false, RequeueAfter: 0}, err
 }
 
@@ -113,5 +134,6 @@
 		Owns(&appsv1.Deployment{}).
 		Owns(&networkingv1.Ingress{}).
 		Owns(&corev1.Service{}).
+		WithEventFilter(predicate.GenerationChangedPredicate{}).
 		Complete(r)
 }

internal/controller/asset_deployment.go

@@ -22,9 +22,9 @@ func (r *AssetReconciler) ReconcileDeployment(log logr.Logger) (bool, error) {
 	}
 	dep := appsv1.Deployment{
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      "asset",
+			Name:      AssetDeploymentName,
 			Namespace: r.NamespacedName.Namespace,
-			Labels:    getAppLabels("asset"),
+			Labels:    getAppLabels(AssetDeploymentName),
 		},
 	}
 	_, err := controllerutil.CreateOrUpdate(r.Context, r.Client, &dep, func() error {
@@ -50,7 +50,7 @@ func (r *AssetReconciler) updateDeployment(dep *appsv1.Deployment, asset *terano
 }
 
 func defaultAssetDeploymentSpec() *appsv1.DeploymentSpec {
-	labels := getAppLabels("asset")
+	labels := getAppLabels(AssetDeploymentName)
 	envFrom := []corev1.EnvFromSource{}
 	env := []corev1.EnvVar{
 		{
@@ -59,7 +59,7 @@ func defaultAssetDeploymentSpec() *appsv1.DeploymentSpec {
 		},
 	}
 	return &appsv1.DeploymentSpec{
-		Replicas: ptr.To(int32(2)),
+		Replicas: ptr.To(int32(DefaultAssetReplicas)),
 		Selector: metav1.SetAsLabelSelector(labels),
 		Strategy: appsv1.DeploymentStrategy{
 			Type: appsv1.RecreateDeploymentStrategyType,

internal/controller/blockassembly_deployment.go

@@ -65,7 +65,7 @@ func defaultBlockAssemblyDeploymentSpec() *appsv1.DeploymentSpec {
 		Strategy: appsv1.DeploymentStrategy{
 			Type: appsv1.RecreateDeploymentStrategyType,
 		},
-		Replicas: ptr.To(int32(1)),
+		Replicas: ptr.To(int32(DefaultBlockAssemblyReplicas)),
 		Selector: metav1.SetAsLabelSelector(labels),
 		Template: corev1.PodTemplateSpec{
 			ObjectMeta: metav1.ObjectMeta{

internal/controller/blockchain_controller.go

@@ -123,6 +123,12 @@ func (r *BlockchainReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 				// FSM is disabled, skip FSM reconciliation
 				return ctrl.Result{}, nil
 			}
+
+			if isConnectionError(err) {
+				r.Log.Info("FSM monitoring unavailable - operator may be running out-of-cluster. FSM state cannot be retrieved when .svc.cluster.local DNS is not resolvable", "error", err.Error())
+				// Don't requeue aggressively if we know we're out-of-cluster
+				return ctrl.Result{Requeue: true, RequeueAfter: time.Minute * 5}, nil
+			}
 			r.Log.Error(err, "unable to get FSM state, trying again in a minute")
 			return ctrl.Result{Requeue: true, RequeueAfter: time.Minute}, nil
 		}

internal/controller/blockchain_deployment.go

@@ -60,7 +60,7 @@ func defaultBlockchainDeploymentSpec() *appsv1.DeploymentSpec {
 	}
 	// TODO: set a default
 	return &appsv1.DeploymentSpec{
-		Replicas: ptr.To(int32(1)),
+		Replicas: ptr.To(int32(DefaultBlockchainReplicas)),
 		Selector: metav1.SetAsLabelSelector(podLabels),
 		Strategy: appsv1.DeploymentStrategy{
 			Type: appsv1.RecreateDeploymentStrategyType,

internal/controller/blockchain_fsm.go

@@ -3,6 +3,7 @@ package controller
 import (
 	"errors"
 	"fmt"
+	"strings"
 
 	"github.com/bsv-blockchain/teranode/services/blockchain"
 	"github.com/bsv-blockchain/teranode/settings"
@@ -90,3 +91,17 @@ func (r *BlockchainReconciler) ReconcileState(state blockchain.FSMStateType) err
 	}
 	return nil
 }
+
+// isConnectionError checks if the error is a gRPC connection/unavailability error
+// This typically happens when the operator runs out-of-cluster and cannot resolve
+// Kubernetes internal DNS names like *.svc.cluster.local
+func isConnectionError(err error) bool {
+	if err == nil {
+		return false
+	}
+	errMsg := err.Error()
+	return strings.Contains(errMsg, "Unavailable") ||
+		strings.Contains(errMsg, "no children to pick from") ||
+		strings.Contains(errMsg, "connection refused") ||
+		strings.Contains(errMsg, "no such host")
+}