feat: rewrite weka_runtime.py in Go (phase 1 - structure, deployment, basic modes)

.dagger/src/apps/operator.py

@@ -59,6 +59,39 @@ async def publish_operator(src: Directory, sock: Socket, repository: str, versio
     )
 
 
+async def publish_pod_runtime(src: Directory, sock: Socket, repository: str, version: str = "",
+                              registry_username: Optional[Secret] = None,
+                              registry_password: Optional[Secret] = None) -> str:
+    """Build and publish multi-arch weka-pod-runtime image (linux/amd64 + linux/arm64)."""
+    version = await _calc_operator_version(src, version)
+
+    pod_runtime_repo = repository.replace("weka-operator", "weka-pod-runtime")
+
+    async def build_variant(p: str) -> Container:
+        target_os, target_arch = p.split("/")
+        binary = await build_go(src, sock, cache_deps=False,
+                                program_path="cmd/weka-pod-runtime/main.go",
+                                target_os=target_os, target_arch=target_arch)
+        return (
+            dag.container(platform=dagger.Platform(p))
+            .with_file("/weka-pod-runtime", binary.file("/out-binary"))
+        )
+
+    variants = await asyncio.gather(*[build_variant(p) for p in PLATFORMS])
+
+    publisher = dag.container()
+    if registry_username and registry_password:
+        registry_host = pod_runtime_repo.split("/")[0]
+        username_str = await registry_username.plaintext()
+        publisher = publisher.with_registry_auth(registry_host, username_str, registry_password)
+
+    ref = await publisher.publish(
+        f"{pod_runtime_repo}:{version}",
+        platform_variants=list(variants),
+    )
+    return ref.split("@")[0]
+
+
 async def publish_operator_helm_chart(src: Directory, sock: Socket, repository: str, version: str = "",
                                       gh_token: Optional[Secret] = None,
                                       helm_username: Optional[Secret] = None,

.dagger/src/operator_flows/main.py

@@ -1,3 +1,4 @@
+import asyncio
 import re
 import logging
 import random
@@ -65,15 +66,15 @@ async def build_to_registry(self,
             helm_username: Optional username for helm registry login
             helm_password: Optional password for helm registry login
         """
-        from apps.operator import publish_operator, publish_operator_helm_chart
-        _ = await publish_operator(operator, sock,
-                                   repository=operator_repo,
-                                   )
-        operator_helm = await publish_operator_helm_chart(operator, sock,
-                                                          repository=helm_repo,
-                                                          helm_username=helm_username,
-                                                          helm_password=helm_password,
-                                                          )
+        from apps.operator import publish_operator, publish_operator_helm_chart, publish_pod_runtime
+        _, operator_helm, _ = await asyncio.gather(
+            publish_operator(operator, sock, repository=operator_repo),
+            publish_operator_helm_chart(operator, sock,
+                                        repository=helm_repo,
+                                        helm_username=helm_username,
+                                        helm_password=helm_password),
+            publish_pod_runtime(operator, sock, repository=operator_repo),
+        )
         return operator_helm
 
     @function
@@ -125,41 +126,41 @@ async def publish_operator_and_get_versions(
         helm_password: Optional[dagger.Secret] = None,
     ) -> List[str]:
         """
-        Build, publish operator and return image versions.
-
-        Args:
-            operator: Operator source directory
-            sock: Docker socket
-            operator_repo: Fully qualified operator image repository (e.g. quay.io/weka.io/weka-operator)
-            helm_repo: Fully qualified Helm chart OCI repository (e.g. quay.io/weka.io/helm)
-            gh_token: Optional GitHub token for private dependencies
-            helm_username: Optional Helm registry username
-            helm_password: Optional Helm registry password
+        Build and publish all images concurrently, return their refs.
 
         Returns:
-            Tuple of (operator_image, operator_helm_image)
+            [operator_image, operator_helm_image, pod_runtime_image]
         """
-        from apps.operator import publish_operator, publish_operator_helm_chart
-
-        operator_image_with_hash = await publish_operator(
-            operator, sock,
-            repository=operator_repo,
-            gh_token=gh_token,
-            registry_username=helm_username,
-            registry_password=helm_password,
-        )
-
-        operator_helm_image = await publish_operator_helm_chart(
-            operator, sock,
-            repository=helm_repo,
-            gh_token=gh_token,
-            helm_username=helm_username,
-            helm_password=helm_password,
+        from apps.operator import publish_operator, publish_operator_helm_chart, publish_pod_runtime
+
+        operator_image_with_hash, operator_helm_image, pod_runtime_image = await asyncio.gather(
+            publish_operator(
+                operator, sock,
+                repository=operator_repo,
+                gh_token=gh_token,
+                registry_username=helm_username,
+                registry_password=helm_password,
+            ),
+            publish_operator_helm_chart(
+                operator, sock,
+                repository=helm_repo,
+                gh_token=gh_token,
+                helm_username=helm_username,
+                helm_password=helm_password,
+            ),
+            publish_pod_runtime(
+                operator, sock,
+                repository=operator_repo,
+                registry_username=helm_username,
+                registry_password=helm_password,
+            ),
         )
 
-        operator_image = operator_image_with_hash.split("@")[0]
-
-        return operator_image, operator_helm_image
+        return [
+            operator_image_with_hash.split("@")[0],
+            operator_helm_image,
+            pod_runtime_image,
+        ]
 
 
     @function
@@ -386,7 +387,7 @@ async def run_upgrade_extended_test(
             upgrade_test_container = upgrade_test_container.with_env_variable("GITHUB_SHA", gh_sha)
 
         if not operator_image or not operator_helm_image:
-            operator_image, operator_helm_image = await self.publish_operator_and_get_versions(
+            operator_image, operator_helm_image, _ = await self.publish_operator_and_get_versions(
                 operator, sock, operator_repo=operator_repo, helm_repo=helm_repo, gh_token=current_gh_token,
                 helm_username=helm_username, helm_password=helm_password,
             )
@@ -762,7 +763,7 @@ async def run_ocp_clients_only_test(
         )
 
         if not operator_image or not operator_helm_image:
-            operator_image, operator_helm_image = await self.publish_operator_and_get_versions(
+            operator_image, operator_helm_image, _ = await self.publish_operator_and_get_versions(
                 operator, sock, operator_repo=operator_repo, helm_repo=helm_repo, gh_token=current_gh_token,
             )
 

charts/weka-operator/resources/weka_runtime.py

@@ -296,7 +296,7 @@ class FeaturesFlags:
     traces_override_in_slash_traces: Union[bool, int] = 1
     supports_binding_to_not_all_interfaces: Union[bool, int] = 2
     agent_validate_60_ports_per_container: Union[bool, int] = 3
-    allow_per_container_driver_interfaces: Union[bool, int] = 4
+    allow_per_container_driver_interface: Union[bool, int] = 4
     weka_get_copy_local_driver_files: Union[bool, int] = 5
     driver_supports_auto_drain: Union[bool, int] = 6
     ssd_proxy_iommu_support: Union[bool, int] = 7

charts/weka-operator/templates/manager.yaml

@@ -139,6 +139,10 @@ spec:
             - name: WEKA_OTEL_PACKAGES_INSTALLER_IMAGE
               value: "{{ .Values.otelPackagesInstallerImage }}"
             {{- end }}
+            - name: WEKA_POD_RUNTIME_IMAGE
+              value: "{{ .Values.wekaPodRuntime.image.repository }}:{{if not (empty .Values.wekaPodRuntime.image.tag)}}{{ .Values.wekaPodRuntime.image.tag }}{{else}}{{ .Chart.Version }}{{end}}"
+            - name: WEKA_USE_PYTHON_FALLBACK
+              value: "{{ .Values.wekaPodRuntime.usePythonFallback }}"
             - name: WEKA_OCP_PULL_SECRET
               value: "{{ .Values.ocpCompatibility.driverToolkitSecretName }}"
             - name: WEKA_OCP_TOOLKIT_IMAGE_BASE_URL

charts/weka-operator/values.yaml

@@ -10,6 +10,14 @@ image:
   repository: quay.io/weka.io/weka-operator
   # Override only, default equals to the chart version
   tag: ""
+
+wekaPodRuntime:
+  image:
+    repository: quay.io/weka.io/weka-pod-runtime
+    tag: ""               # defaults to .Chart.AppVersion
+    pullPolicy: IfNotPresent
+  usePythonFallback: false
+
 kubeProxyImage: "quay.io/brancz/kube-rbac-proxy:v0.21.0"
 
 builderImages:

cmd/weka-pod-runtime/main.go

@@ -0,0 +1,44 @@
+package main
+
+import (
+	"context"
+	"os"
+	"os/signal"
+	"syscall"
+
+	"github.com/weka/go-weka-observability/instrumentation"
+	obslogger "github.com/weka/go-weka-observability/logger"
+	"github.com/weka/weka-operator/internal/runtime/config"
+	"github.com/weka/weka-operator/internal/runtime/modes"
+)
+
+func main() {
+	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
+
+	cfg := config.Load()
+
+	logger := obslogger.NewZerologrWithLoggerNameInsteadCaller()
+	ctx = obslogger.ContextWithLogr(ctx, logger)
+
+	shutdown, err := instrumentation.SetupOTelSDKWithOptions(ctx, "weka-pod-runtime", cfg.Version, logger)
+	if err != nil {
+		// observability is non-critical, log and continue
+		logger.Info("failed to set up OTel SDK", "err", err)
+	}
+	if err := modes.Run(ctx, cfg); err != nil {
+		logger.Error(err, "mode failed", "mode", cfg.Mode)
+		if shutdown != nil {
+			if shutdownErr := shutdown(ctx); shutdownErr != nil {
+				logger.Info("failed to shutdown OTel", "err", shutdownErr)
+			}
+		}
+		stop()
+		os.Exit(1)
+	}
+	if shutdown != nil {
+		if shutdownErr := shutdown(ctx); shutdownErr != nil {
+			logger.Info("failed to shutdown OTel", "err", shutdownErr)
+		}
+	}
+	stop()
+}

go.mod

@@ -15,6 +15,7 @@ require (
 	github.com/prometheus/common v0.66.1
 	github.com/rs/zerolog v1.34.0
 	github.com/sethvargo/go-password v0.3.1
+	github.com/stretchr/testify v1.11.1
 	github.com/thessem/zap-prettyconsole v0.5.2
 	github.com/weka/go-lib v0.0.0-20250813155110-01f071a2760a
 	github.com/weka/go-steps-engine v0.0.0
@@ -79,6 +80,7 @@ require (
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/mxk/go-flowrate v0.0.0-20140419014527-cca7078d478f // indirect
+	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/prometheus/client_golang v1.23.2 // indirect
 	github.com/prometheus/procfs v0.17.0 // indirect
 	github.com/rogpeppe/go-internal v1.14.1 // indirect

internal/config/env.go

@@ -254,6 +254,9 @@ var Config struct {
 	PodConfigVersion                     string
 	EnablePodConfigCodeVersionRotation   bool
 	AllowRotateNonAnnotatedPodConfigHash bool
+
+	WekaPodRuntimeImage string
+	UsePythonFallback   bool
 }
 
 type NodeAgentRequestsTimeouts struct {
@@ -502,6 +505,9 @@ func ConfigureEnv(ctx context.Context) {
 	// Evicted pod cleanup configuration
 	Config.EvictedPodCleanupEnabled = getBoolEnvOrDefault("EVICTED_POD_CLEANUP_ENABLED", true)
 	Config.EvictedPodCleanupInterval = getDurationEnvOrDefault("EVICTED_POD_CLEANUP_INTERVAL", 2*time.Minute)
+
+	Config.WekaPodRuntimeImage = os.Getenv("WEKA_POD_RUNTIME_IMAGE") // No default - opt-in only
+	Config.UsePythonFallback = getBoolEnvOrDefault("WEKA_USE_PYTHON_FALLBACK", false)
 }
 
 func getEnvOrFail(envKey string) string {

internal/controllers/operations/discover_drives.go

@@ -43,23 +43,9 @@ type DiscoverDrivesOperation struct {
 	tolerations     []corev1.Toleration
 }
 
-type DriveRawInfo struct {
-	SerialId    string `json:"serial_id"`
-	Path        string `json:"path"`
-	IsMounted   bool   `json:"is_mounted"`
-	CapacityGiB int    `json:"capacity_gib"`
-}
-
-type DriveNodeResults struct {
-	Err         error                    `json:"err"`
-	Drives      []domain.DriveInfo       `json:"drives"`
-	RawDrives   []DriveRawInfo           `json:"raw_drives"`
-	ProxyDrives []domain.SharedDriveInfo `json:"proxy_drives,omitempty"` // Signed drives for proxy mode
-}
-
 type DiscoverDrivesResult struct {
-	Err     string                      `json:"err,omitempty"`
-	Results map[string]DriveNodeResults `json:"results"`
+	Err     string                            `json:"err,omitempty"`
+	Results map[string]domain.DriveNodeResults `json:"results"`
 }
 
 func NewDiscoverDrivesOperation(mgr ctrl.Manager, payload *v1alpha1.DiscoverDrivesPayload, ownerRef client.Object, ownerDetails v1alpha1.WekaOwnerDetails, ownerStatus string, successCallback lifecycle.StepFunc, force bool) *DiscoverDrivesOperation { //nolint:gocritic // intentional code pattern, linter suggestion does not apply here
@@ -212,7 +198,7 @@ func processResult(ctx context.Context, containers []*v1alpha1.WekaContainer, sk
 	_, logger := instrumentation.CreateLogSpan(ctx, "ProcessResult")
 	defer logger.End()
 
-	results := make(map[string]DriveNodeResults)
+	results := make(map[string]domain.DriveNodeResults)
 	errorCount := 0
 
 	for _, container := range containers {
@@ -224,13 +210,13 @@ func processResult(ctx context.Context, containers []*v1alpha1.WekaContainer, sk
 			return nil, lifecycle.NewWaitErrorWithDuration(err, time.Second*10)
 		}
 
-		var opResult DriveNodeResults
+		var opResult domain.DriveNodeResults
 		err := json.Unmarshal([]byte(*container.Status.ExecutionResult), &opResult)
 		logger.Info("Processing container result", "container", container.Name, "result", opResult)
 		if err != nil {
-			errs := err.Error()
-			results[string(container.GetNodeAffinity())] = DriveNodeResults{
-				Err: fmt.Errorf("failed to unmarshal execution result: %s", errs),
+			errMsg := fmt.Sprintf("failed to unmarshal execution result: %s", err.Error())
+			results[string(container.GetNodeAffinity())] = domain.DriveNodeResults{
+				Err: &errMsg,
 			}
 			continue
 		}

internal/controllers/operations/sign_drives.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"slices"
 	"strings"
 	"time"
 
@@ -32,11 +33,6 @@ import (
 	"github.com/weka/weka-operator/pkg/util"
 )
 
-type SignedDrivesExtendedPayload struct {
-	weka.SignDrivesPayload
-	ExcludedSerialIds     []string `json:"excludedSerialIds,omitempty"`
-	SsdProxyContainerUuid string   `json:"ssd_proxy_container_uuid,omitempty"`
-}
 
 type SignDrivesOperation struct {
 	client          client.Client
@@ -129,7 +125,7 @@ func (o *SignDrivesOperation) EnsureContainers(ctx context.Context) error {
 	}
 
 	// Create a copy of the original payload to avoid modifying it
-	extendedPayload := SignedDrivesExtendedPayload{
+	extendedPayload := domain.SignedDrivesExtendedPayload{
 		SignDrivesPayload: *o.payload,
 	}
 
@@ -273,12 +269,9 @@ func (o *SignDrivesOperation) EnsureContainers(ctx context.Context) error {
 // processResults (i.e. updateNodeAnnotations). Deleting before this condition is set
 // causes node annotations to never be written.
 func isResultsProcessed(container *weka.WekaContainer) bool {
-	for _, c := range container.Status.Conditions {
-		if c.Type == condition.CondResultsProcessed && c.Status == metav1.ConditionTrue {
-			return true
-		}
-	}
-	return false
+	return slices.ContainsFunc(container.Status.Conditions, func(c metav1.Condition) bool {
+		return c.Type == condition.CondResultsProcessed && c.Status == metav1.ConditionTrue
+	})
 }
 
 func (o *SignDrivesOperation) PollResults(ctx context.Context) error {
@@ -334,12 +327,12 @@ func (o *SignDrivesOperation) GetJsonResult() string {
 		}
 		if nodeResults.Err != nil {
 			if len(errs) < maxErrors {
-				errs = append(errs, nodeResults.Err.Error())
+				errs = append(errs, *nodeResults.Err)
 			}
 		}
 	}
 
-	ret := map[string]interface{}{}
+	ret := map[string]any{}
 	if len(drivesByNode) > 0 {
 		ret["results"] = drivesByNode
 		ret["message"] = fmt.Sprintf("Signed %d drives on %d nodes", total, len(o.results.Results))
@@ -433,7 +426,7 @@ func getAlreadySignedDrives(node *v1.Node) []string {
 	return alreadySignedDrives
 }
 
-func (o *SignDrivesOperation) createInstructions(extendedPayload *SignedDrivesExtendedPayload) (*weka.Instructions, error) {
+func (o *SignDrivesOperation) createInstructions(extendedPayload *domain.SignedDrivesExtendedPayload) (*weka.Instructions, error) {
 	// Marshal the extended payload
 	payloadBytes, err := json.Marshal(extendedPayload)
 	if err != nil {