Merge branch 'master' into 3723-PersistentVolumeClaim

WORKSPACE

@@ -96,7 +96,7 @@ maven_install(
         "org.glassfish.jersey.media:jersey-media-multipart:" + jersey_version,
         "org.glassfish.jersey.containers:jersey-container-servlet:" + jersey_version,
         "org.apache.distributedlog:distributedlog-core:" + distributedlog_version,
-        "io.netty:netty-all:4.1.50.Final",
+        "io.netty:netty-all:4.1.70.Final",
         "aopalliance:aopalliance:1.0",
         "org.roaringbitmap:RoaringBitmap:0.6.51",
         "com.google.guava:guava:23.6-jre",

deploy/kubernetes/general/zookeeper.yaml

@@ -21,7 +21,7 @@
 
 ## Define a disruption budget to ensure there are at least
 ## 2 ZK servers running all the time
-apiVersion: policy/v1beta1
+apiVersion: policy/v1
 kind: PodDisruptionBudget
 metadata:
   name: zk-budget

deploy/kubernetes/helm/templates/tools.yaml

@@ -297,6 +297,12 @@ rules:
   - patch
   - update
   - watch
+- apiGroups:
+  - ""
+  resources:
+  - pods
+  verbs:
+  - deletecollection
 - apiGroups:
   - ""
   resources:

deploy/kubernetes/helm/templates/zookeeper.yaml

@@ -18,7 +18,7 @@
 {{- $platform := .Values.platform }}
 {{- $zkReplicas := .Values.zkReplicas }}
 
-apiVersion: policy/v1beta1
+apiVersion: policy/v1
 kind: PodDisruptionBudget
 metadata:
   name: {{ .Release.Name }}-zookeeper

heron/schedulers/src/java/org/apache/heron/scheduler/kubernetes/V1Controller.java

@@ -178,12 +178,23 @@ boolean killTopology() {
     return true;
   }
 
+  /**
+   * Restarts a topology by deleting the Pods associated with it using <code>Selector Labels</code>.
+   * @param shardId Not used but required because of interface.
+   * @return Indicator of successful submission of restart request to Kubernetes cluster.
+   */
   @Override
   boolean restart(int shardId) {
-    final String message = "Restarting the whole topology is not supported yet. "
-        + "Please kill and resubmit the topology.";
-    LOG.log(Level.SEVERE, message);
-    return false;
+    try {
+      coreClient.deleteCollectionNamespacedPod(getNamespace(), null, null, null, null,
+          0, createTopologySelectorLabels(), null, null, null, null,
+          null, null, null);
+      LOG.log(Level.WARNING, String.format("Restarting topology '%s'...", getTopologyName()));
+    } catch (ApiException e) {
+      LOG.log(Level.SEVERE, String.format("Failed to restart topology '%s'...", getTopologyName()));
+      return false;
+    }
+    return true;
   }
 
   @Override
@@ -1074,4 +1085,14 @@ protected static Map<String, String> getPersistentVolumeClaimLabels(String topol
       }
     };
   }
+
+   /*
+   * Generates the <code>Selector</code> match labels with which resources in this topology can be found.
+   * @return A label of the form <code>app=heron,topology=topology-name</code>.
+   */
+  private String createTopologySelectorLabels() {
+    return String.format("%s=%s,%s=%s",
+        KubernetesConstants.LABEL_APP, KubernetesConstants.LABEL_APP_VALUE,
+        KubernetesConstants.LABEL_TOPOLOGY, getTopologyName());
+  }
 }

integration_test/src/python/http_server/BUILD

@@ -8,6 +8,7 @@ pex_binary(
     main = "main.py",
     reqs = [
         "tornado==4.5.3",
+        "werkzeug==2.0.2",
     ],
     deps = [
         "//heron/common/src/python:common-py",

integration_test/src/python/http_server/main.py

@@ -21,6 +21,7 @@
 import tornado.ioloop
 import tornado.escape
 import tornado.web
+from werkzeug.utils import secure_filename
 
 from heron.common.src.python.utils import log
 
@@ -32,7 +33,7 @@ def get(self):
 
 class FileHandler(tornado.web.RequestHandler):
   def get(self, fileName):
-    jsonFilePath = RESULTS_DIRECTORY + "/" + fileName + ".json"
+    jsonFilePath = RESULTS_DIRECTORY + "/" + secure_filename(fileName) + ".json"
 
     if not os.path.exists(jsonFilePath):
       self.clear()
@@ -46,7 +47,7 @@ def get(self, fileName):
       self.write(data)
 
   def post(self, fileName):
-    jsonFilePath = RESULTS_DIRECTORY + "/" + fileName + ".json"
+    jsonFilePath = RESULTS_DIRECTORY + "/" + secure_filename(fileName) + ".json"
 
     #Overwrites the existing file
     with open(jsonFilePath, "w") as jsonFile: