Bump marquez to 0.12.2

Signed-off-by: wslulciuc <[email protected]>

Author: wslulciuc

Dockerfile

@@ -1,4 +1,4 @@
-FROM openjdk:11-jdk AS base
+FROM adoptopenjdk/openjdk11:alpine AS base
 WORKDIR /usr/src/app
 COPY gradle gradle
 COPY gradle.properties gradle.properties
@@ -14,11 +14,11 @@ COPY api/build.gradle ./api/build.gradle
 COPY clients/java ./clients/java
 RUN ./gradlew --no-daemon :api:shadowJar
 
-FROM openjdk:11-jre
-RUN apt-get update && apt-get install -y --no-install-recommends postgresql-client
+FROM adoptopenjdk/openjdk11:alpine-jre
+RUN apk update && apk add --virtual postgresql-client bash coreutils
 WORKDIR /usr/src/app
 COPY --from=build /usr/src/app/api/build/libs/marquez-*.jar /usr/src/app
 COPY marquez.dev.yml marquez.dev.yml
 COPY docker/entrypoint.sh entrypoint.sh
 EXPOSE 5000 5001
-ENTRYPOINT ["/usr/src/app/entrypoint.sh"]
+ENTRYPOINT ["/usr/src/app/entrypoint.sh"]

api/gradle.properties

@@ -1 +1 @@
-version=0.12.0
+version=0.12.2

clients/gradle.properties

@@ -0,0 +1 @@
+version=0.12.2

clients/java/gradle.properties

@@ -10,4 +10,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-VERSION = '0.12.0'
+VERSION = '0.12.2'

clients/python/marquez_client/version.py

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 0.12.0
+current_version = 0.12.2
 
 [metadata]
 description-file = README.md

clients/python/setup.cfg

@@ -34,18 +34,21 @@ fi
 
 echo "Building images (tag: ${version})..."
 
-# Build and tag app image
-docker build --no-cache --tag "marquez:${version}" .
-docker tag 'marquez' 'marquez:latest'
-
-# Build and tag web image
-docker build --no-cache --tag "marquez-web:${version}" .
-docker tag 'marquez-web' 'marquez-web:latest'
-
-# Push images to Docker Hub
-docker push "marquez:${version}"
-docker push 'marquez:latest'
-docker push "marquez-web:${version}"
-docker push 'marquez-web:latest'
+# Build, tag and push app image
+docker build --no-cache --tag "${ORG}/marquez:${version}" .
+docker tag "${ORG}/marquez:${version}" "${ORG}/marquez:latest"
+
+docker push "${ORG}/marquez:${version}"
+docker push "${ORG}/marquez:latest"
+
+# Change working directory to web module
+cd "${project_root}"/web
+
+# Build, tag and push web image
+docker build --no-cache --tag "${ORG}/marquez-web:${version}" .
+docker tag "${ORG}/marquez-web:${version}" "${ORG}/marquez-web:latest"
+
+docker push "${ORG}/marquez-web:${version}"
+docker push "${ORG}/marquez-web:latest"
 
 echo "DONE!"

docker/build-and-push.sh

@@ -1,4 +1,4 @@
 org.gradle.internal.http.connectionTimeout=60000
 org.gradle.internal.http.socketTimeout=60000
 
-version=0.11.4-rc.4
+version=0.12.2

docs/openapi.html

@@ -10,12 +10,11 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import logging
 from uuid import uuid4
 
 import airflow.models
 import time
-from airflow import LoggingMixin
+
 from airflow.contrib.operators.bigquery_operator import BigQueryOperator
 from airflow.operators.postgres_operator import PostgresOperator
 from airflow.utils.state import State
@@ -30,39 +29,41 @@
     DagUtils
 )
 
+# Handling of import of different airflow versions
+from airflow.version import version as AIRFLOW_VERSION
+from pkg_resources import parse_version
+if parse_version(AIRFLOW_VERSION) >= parse_version("1.10.11"):
+    from airflow import LoggingMixin
+else:
+    # Corrects path of import for Airflow versions below 1.10.11
+    from airflow.utils.log.logging_mixin import LoggingMixin
+
 from marquez_airflow.marquez import Marquez
 
-log = logging.getLogger(__name__)
 
+_MARQUEZ = Marquez()
+
+# TODO: Manually define operator->extractor mappings for now,
+# but we'll want to encapsulate this logic in an 'Extractors' class
+# with more convenient methods (ex: 'Extractors.extractor_for_task()')
+_EXTRACTORS = {
+    PostgresOperator: PostgresExtractor,
+    BigQueryOperator: BigQueryExtractor
+    # Append new extractors here
+}
 
-class DAG(airflow.models.DAG, LoggingMixin):
-    _job_id_mapping = None
-    _marquez = None
 
+class DAG(airflow.models.DAG, LoggingMixin):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
 
-        self._job_id_mapping = JobIdMapping()
-        self._marquez = Marquez()
-        # TODO: Manually define operator->extractor mappings for now,
-        # but we'll want to encapsulate this logic in an 'Extractors' class
-        # with more convenient methods (ex: 'Extractors.extractor_for_task()')
-        self._extractors = {
-            PostgresOperator: PostgresExtractor,
-            BigQueryOperator: BigQueryExtractor
-            # Append new extractors here
-        }
-        self.log.debug(
-            f"DAG successfully created with extractors: {self._extractors}"
-        )
-
     def create_dagrun(self, *args, **kwargs):
         # run Airflow's create_dagrun() first
         dagrun = super(DAG, self).create_dagrun(*args, **kwargs)
 
         create_dag_start_ms = self._now_ms()
         try:
-            self._marquez.create_namespace()
+            _MARQUEZ.create_namespace()
             self._register_dagrun(
                 dagrun,
                 DagUtils.get_execution_date(**kwargs),
@@ -83,10 +84,10 @@ def _register_dagrun(self, dagrun, execution_date, run_args):
             t = self._now_ms()
             try:
                 steps = self._extract_metadata(dagrun, task)
-                [self._marquez.create_job(
+                [_MARQUEZ.create_job(
                     step, self._get_location(task), self.description)
                     for step in steps]
-                marquez_jobrun_ids = [self._marquez.create_run(
+                marquez_jobrun_ids = [_MARQUEZ.create_run(
                     self.new_run_id(),
                     step,
                     run_args,
@@ -95,7 +96,7 @@ def _register_dagrun(self, dagrun, execution_date, run_args):
                         execution_date,
                         self.following_schedule(execution_date))
                 ) for step in steps]
-                self._job_id_mapping.set(
+                JobIdMapping.set(
                     self._marquez_job_name(self.dag_id, task.task_id),
                     dagrun.run_id,
                     marquez_jobrun_ids)
@@ -110,7 +111,7 @@ def handle_callback(self, *args, **kwargs):
         try:
             dagrun = args[0]
             self.log.debug(f"handle_callback() dagrun : {dagrun}")
-            self._marquez.create_namespace()
+            _MARQUEZ.create_namespace()
             self._report_task_instances(
                 dagrun,
                 DagUtils.get_run_args(**kwargs),
@@ -137,17 +138,17 @@ def _report_task_instances(self, dagrun, run_args, session):
 
     def _report_task_instance(self, ti, dagrun, run_args, session):
         task = self.get_task(ti.task_id)
-        run_ids = self._job_id_mapping.pop(
+        run_ids = JobIdMapping.pop(
             self._marquez_job_name_from_ti(ti), dagrun.run_id, session)
         steps = self._extract_metadata(dagrun, task, ti)
 
         # Note: run_ids could be missing if it was removed from airflow
         # or the job could not be registered.
         if not run_ids:
-            [self._marquez.create_job(
+            [_MARQUEZ.create_job(
                 step, self._get_location(task), self.description)
-             for step in steps]
-            run_ids = [self._marquez.create_run(
+                for step in steps]
+            run_ids = [_MARQUEZ.create_run(
                 self.new_run_id(),
                 step,
                 run_args,
@@ -159,30 +160,30 @@ def _report_task_instance(self, ti, dagrun, run_args, session):
 
         for step in steps:
             for run_id in run_ids:
-                self._marquez.create_job(
+                _MARQUEZ.create_job(
                     step, self._get_location(task), self.description,
                     ti.state, run_id)
-                self._marquez.start_run(
+                _MARQUEZ.start_run(
                     run_id,
                     DagUtils.to_iso_8601(ti.start_date))
 
                 self.log.debug(f'Setting task state: {ti.state}'
                                f' for {ti.task_id}')
                 if ti.state in {State.SUCCESS, State.SKIPPED}:
-                    self._marquez.complete_run(
+                    _MARQUEZ.complete_run(
                         run_id,
                         DagUtils.to_iso_8601(ti.end_date))
                 else:
-                    self._marquez.fail_run(
+                    _MARQUEZ.fail_run(
                         run_id,
                         DagUtils.to_iso_8601(ti.end_date))
 
     def _extract_metadata(self, dagrun, task, ti=None):
         extractor = self._get_extractor(task)
         task_info = f'task_type={task.__class__.__name__} ' \
-            f'airflow_dag_id={self.dag_id} ' \
-            f'task_id={task.task_id} ' \
-            f'airflow_run_id={dagrun.run_id} '
+                    f'airflow_dag_id={self.dag_id} ' \
+                    f'task_id={task.task_id} ' \
+                    f'airflow_run_id={dagrun.run_id} '
         if extractor:
             try:
                 self.log.debug(
@@ -216,13 +217,13 @@ def _extract(self, extractor, task, ti):
         return extractor(task).extract()
 
     def _get_extractor(self, task):
-        extractor = self._extractors.get(task.__class__)
-        log.debug(f'extractor for {task.__class__} is {extractor}')
+        extractor = _EXTRACTORS.get(task.__class__)
+        self.log.debug(f'extractor for {task.__class__} is {extractor}')
         return extractor
 
     def _timed_log_message(self, start_time):
         return f'airflow_dag_id={self.dag_id} ' \
-            f'duration_ms={(self._now_ms() - start_time)}'
+               f'duration_ms={(self._now_ms() - start_time)}'
 
     def new_run_id(self) -> str:
         return str(uuid4())
@@ -239,8 +240,6 @@ def _get_location(task):
             else:
                 return get_location(task.dag.fileloc)
         except Exception:
-            log.warning(f"Failed to get location for task '{task.task_id}'.",
-                        exc_info=True)
             return None
 
     @staticmethod