fix pipeline reconciler namespacing (#108)

Author: jogrogan

hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sApi.java

@@ -108,7 +108,7 @@ public void update(T obj) throws SQLException {
   }
 
   public void updateStatus(T obj, Object status) throws SQLException {
-    if (!endpoint.clusterScoped()) {
+    if (obj.getMetadata().getNamespace() == null && !endpoint.clusterScoped()) {
       obj.getMetadata().namespace(context.namespace());
     }
     KubernetesApiResponse<T> resp = context.<T, U>generic(endpoint).updateStatus(obj, x -> status);

hoptimator-operator/src/main/java/com/linkedin/hoptimator/operator/PipelineOperatorApp.java

@@ -26,9 +26,11 @@
 public class PipelineOperatorApp {
   private static final Logger log = LoggerFactory.getLogger(PipelineOperatorApp.class);
 
+  final String watchNamespace;
   final Properties connectionProperties;
 
-  public PipelineOperatorApp(Properties connectionProperties) {
+  public PipelineOperatorApp(String watchNamespace, Properties connectionProperties) {
+    this.watchNamespace = watchNamespace;
     this.connectionProperties = connectionProperties;
   }
 
@@ -55,18 +57,14 @@ public static void main(String[] args) throws Exception {
     }
 
     String watchNamespaceInput = cmd.getOptionValue("watch", "");
-
-    Properties props = new Properties();
-    props.put("k8s.namespace", watchNamespaceInput);
-
-    new PipelineOperatorApp(props).run();
+    new PipelineOperatorApp(watchNamespaceInput, new Properties()).run();
   }
 
   public void run() throws Exception {
     K8sContext context = new K8sContext(connectionProperties);
 
     // register informers
-    context.registerInformer(K8sApiEndpoints.PIPELINES, Duration.ofMinutes(5), context.namespace());
+    context.registerInformer(K8sApiEndpoints.PIPELINES, Duration.ofMinutes(5), watchNamespace);
 
     List<Controller> controllers = new ArrayList<>();
     // TODO: add additional controllers from ControllerProvider SPI

hoptimator-operator/src/main/java/com/linkedin/hoptimator/operator/pipeline/PipelineReconciler.java

@@ -2,6 +2,7 @@
 
 import java.time.Duration;
 import java.util.Arrays;
+import java.util.Properties;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -42,10 +43,11 @@ private PipelineReconciler(K8sContext context) {
   public Result reconcile(Request request) {
     log.info("Reconciling request {}", request);
     String name = request.getName();
+    String namespace = request.getNamespace();
 
     Result result = new Result(true, pendingRetryDuration());
     try {
-      V1alpha1Pipeline object = pipelineApi.get(name);
+      V1alpha1Pipeline object = pipelineApi.get(namespace, name);
 
       if (object == null) {
         log.info("Object {} deleted. Skipping.", name);