Merge branch 'main' into person/aivanov/contrib

Author: antoniivanov

projects/control-service/projects/pipelines_control_service/src/integration-test/java/com/vmware/taurus/datajobs/it/DataJobTerminationStatusIT.java

@@ -150,7 +150,9 @@ public void testDataJobTerminationStatus(String jobName, String teamName, String
     System.out.println(match.get().trim());
     assertTrue(
         match.get().trim().endsWith("0.0"),
-        "The value of the taurus_datajob_termination_status metrics does not match");
+        "The value of the taurus_datajob_termination_status metrics does not match. It was actually"
+            + " "
+            + match.get());
 
     // Check the data job execution status
     checkDataJobExecutionStatus(

projects/control-service/projects/pipelines_control_service/src/main/java/com/vmware/taurus/service/KubernetesService.java

@@ -422,7 +422,7 @@ public void createNamespace(String namespaceName) throws ApiException {
             .withMetadata(new V1ObjectMetaBuilder().withName(namespaceName).build())
             .build();
 
-    new CoreV1Api(client).createNamespace(namespaceBody, null, null, null);
+    new CoreV1Api(client).createNamespace(namespaceBody, null, null, null, null);
   }
 
   public void deleteNamespace(String namespaceName) throws ApiException {
@@ -923,7 +923,8 @@ public void createV1beta1CronJob(
             jobLabels,
             imagePullSecrets);
     V1beta1CronJob nsJob =
-        new BatchV1beta1Api(client).createNamespacedCronJob(namespace, cronJob, null, null, null);
+        new BatchV1beta1Api(client)
+            .createNamespacedCronJob(namespace, cronJob, null, null, null, null);
     log.debug("Created k8s cron job: {}", nsJob);
     log.debug(
         "Created k8s cron job name: {}, uid:{}, link:{}",
@@ -967,7 +968,7 @@ public void createV1CronJob(
             jobLabels,
             imagePullSecrets);
     V1CronJob nsJob =
-        new BatchV1Api(client).createNamespacedCronJob(namespace, cronJob, null, null, null);
+        new BatchV1Api(client).createNamespacedCronJob(namespace, cronJob, null, null, null, null);
     log.debug("Created k8s cron job: {}", nsJob);
     log.debug(
         "Created k8s cron job name: {}, uid:{}, link:{}",
@@ -1119,7 +1120,7 @@ public void updateV1beta1CronJob(
             imagePullSecrets);
     V1beta1CronJob nsJob =
         new BatchV1beta1Api(client)
-            .replaceNamespacedCronJob(name, namespace, cronJob, null, null, null);
+            .replaceNamespacedCronJob(name, namespace, cronJob, null, null, null, null);
     log.debug(
         "Updated k8s cron job status for name:{}, image:{}, uid:{}, link:{}",
         name,
@@ -1160,7 +1161,8 @@ public void updateV1CronJob(
             jobLabels,
             imagePullSecrets);
     V1CronJob nsJob =
-        new BatchV1Api(client).replaceNamespacedCronJob(name, namespace, cronJob, null, null, null);
+        new BatchV1Api(client)
+            .replaceNamespacedCronJob(name, namespace, cronJob, null, null, null, null);
     log.debug(
         "Updated k8s cron job status for name:{}, image:{}, uid:{}, link:{}",
         name,
@@ -1258,7 +1260,8 @@ void createNewJob(
             .withSpec(spec)
             .build();
 
-    V1Job nsJob = new BatchV1Api(client).createNamespacedJob(namespace, job, null, null, null);
+    V1Job nsJob =
+        new BatchV1Api(client).createNamespacedJob(namespace, job, null, null, null, null);
     log.debug("Created k8s job: {}", nsJob);
     log.debug(
         "Created k8s job name: {}, uid:{}, link:{}",
@@ -2314,12 +2317,12 @@ public void saveSecretData(String name, Map<String, byte[]> data) throws ApiExce
 
     V1Secret nsSecret;
     try {
-      nsSecret = api.replaceNamespacedSecret(name, this.namespace, secret, null, null, null);
+      nsSecret = api.replaceNamespacedSecret(name, this.namespace, secret, null, null, null, null);
     } catch (ApiException e) {
       log.warn("Error while trying to save K8S secret", e);
       if (e.getCode() == 404) {
         log.debug("Secret {} does not exist. Creating ...", name);
-        nsSecret = api.createNamespacedSecret(this.namespace, secret, null, null, null);
+        nsSecret = api.createNamespacedSecret(this.namespace, secret, null, null, null, null);
       } else {
         log.error("Failed to save k8s secret: {}", name);
         throw e;

projects/control-service/projects/versions-of-external-dependencies.gradle

@@ -28,8 +28,8 @@ project.ext {
             'org.zeroturnaround:zt-exec'                                         : 'org.zeroturnaround:zt-exec:1.12',
             'javax.mail:mail'                                                    : 'javax.mail:mail:1.4.7',
             'org.ini4j:ini4j'                                                    : 'org.ini4j:ini4j:0.5.4',
-            'io.kubernetes:client-java'                                          : 'io.kubernetes:client-java:14.0.1',
-            'io.kubernetes:client-java-api-fluent'                               : 'io.kubernetes:client-java-api-fluent:14.0.1',
+            'io.kubernetes:client-java'                                          : 'io.kubernetes:client-java:16.0.0',
+            'io.kubernetes:client-java-api-fluent'                               : 'io.kubernetes:client-java-api-fluent:16.0.0',
             'com.graphql-java:graphql-java-spring-boot-starter-webmvc'           : 'com.graphql-java:graphql-java-spring-boot-starter-webmvc:2.0',
             'com.cronutils:cron-utils'                                           : 'com.cronutils:cron-utils:9.1.5',
             'net.lingala.zip4j:zip4j'                                            : 'net.lingala.zip4j:zip4j:2.9.0',

projects/vdk-plugins/README.md

@@ -32,7 +32,7 @@ pip install -U cookiecutter
 Generate a VDK Plugin package project:
 
 ```
-cookiecutter https://github.com/tozkata/cookiecutter-vdk-plugin.git
+cookiecutter https://github.com/tozka/cookiecutter-vdk-plugin.git
 ```
 
 Then

projects/vdk-plugins/vdk-impala/src/vdk/plugin/impala/impala_lineage_plugin.py

@@ -145,11 +145,11 @@ def _parse_inputs_outputs(query_profile: str) -> Tuple[list, str]:
         inputs = []
         output = None
         for line in query_profile.splitlines():
-            match = re.search(r"(?<=SCAN HDFS \[)\S*(?=,)", line)
+            match = re.search(r"(?<=SCAN HDFS \[)[\w\.]*", line)
             if match:
                 inputs.append(match.group(0))
             else:
-                match = re.search(r"(?<=WRITE TO HDFS \[)\S*(?=,)", line)
+                match = re.search(r"(?<=WRITE TO HDFS \[)[\w\.]*", line)
                 if match:
                     output = match.group(0)
         return list(set(inputs)), output

projects/vdk-plugins/vdk-impala/tests/impala_lineage_plugin_test.py

@@ -73,6 +73,19 @@ def test_is_query_have_lineage(self):
             )
         )
 
+    def test_parsing_query_profile(self):
+        inputs = {"database_a.table1", "database_b.table2"}
+        output = "database_c.table1"
+        result = ImpalaLineagePlugin._parse_inputs_outputs(
+            """
+            00:SCAN HDFS [database_a.table1, RANDOM]
+            01:SCAN HDFS [database_b.table2 w, RANDOM]
+            03: WRITE TO HDFS [database_c.table1, OVERWRITE=true]
+            """
+        )
+        self.assertEqual(set(result[0]), inputs)
+        self.assertEqual(result[1], output)
+
 
 if __name__ == "__main__":
     unittest.main()

projects/vdk-plugins/vdk-ingest-http/src/vdk/plugin/ingest_http/ingest_http_plugin.py

@@ -58,7 +58,7 @@ def vdk_configure(config_builder: ConfigurationBuilder) -> None:
     )
     config_builder.add(
         key="INGEST_OVER_HTTP_RETRY_BACKOFF_FACTOR",
-        default_value=0,
+        default_value=0.0,
         description="A backoff factor to apply between attempts after the second try "
         "(most errors are resolved immediately by a second try without a "
         "delay). urllib3 will sleep for:: "

projects/vdk-plugins/vdk-ingest-http/tests/functional/test_http_ingest.py

@@ -38,3 +38,26 @@ def test_http_ingestion(httpserver: PluginHTTPServer):
 
         # a single record/row is 100 bytes with 100 records would result is 10 batches of 1000 bytes
         assert len(httpserver.log) == 10
+
+
+def test_ingestion_retry(httpserver: PluginHTTPServer):
+    httpserver.expect_request(uri="/ingest").respond_with_response(Response(status=502))
+
+    with mock.patch.dict(
+        os.environ,
+        {
+            "VDK_INGEST_METHOD_DEFAULT": "http",
+            "VDK_INGEST_TARGET_DEFAULT": httpserver.url_for("/ingest"),
+            "VDK_INGEST_OVER_HTTP_RETRY_TOTAL": "3",
+            "VDK_INGEST_OVER_HTTP_RETRY_BACKOFF_FACTOR": "0.01",  # very short retry backoff interval
+            "VDK_INGEST_OVER_HTTP_RETRY_STATUS_FORCELIST": "502",
+            "VDK_INGESTER_PAYLOAD_SIZE_BYTES_THRESHOLD": "20000",
+        },
+    ):
+        runner = CliEntryBasedTestRunner(ingest_http_plugin)
+
+        result: Result = runner.invoke(["run", job_path("ingest-job")])
+        cli_assert_equal(1, result)
+
+        # one "regular" request + 3 retries
+        assert len(httpserver.log) == 4