Page MenuHomePhabricator

Initial k8s operator and split out of shared functionality
ClosedPublic

Authored by natekupp on Aug 12 2019, 9:44 PM.

Details

Summary
  1. Moving shit around: move our Python and Docker operator into two files
  2. Vendor Airflow k8s operator
  3. Add James' DagsterKubernetesPodOperator

As a follow-up, will need to figure out how we add tests for (3).

Test Plan

tested manually in local airflow + minikube with:

'''
The airflow DAG scaffold for dagster_examples.toys.optional_outputs.optional_outputs

Note that this docstring must contain the strings "airflow" and "DAG" for
Airflow to properly detect it as a DAG
See: http://bit.ly/307VMum
'''
import datetime
import yaml

from dagster_airflow.factory import make_airflow_dag_kubernetized


ENVIRONMENT = '''
loggers:
  console:
    config:
      log_level: DEBUG

solids:
  multiply_the_word:
    inputs:
      word:
        value: bar
    config:
      factor: 2

storage:
  s3:
    config:
      s3_bucket: dagster-airflow-scratch

'''

DEFAULT_ARGS = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime.datetime(2019, 5, 7),
    'email': ['airflow@example.com'],
    'email_on_failure': False,
    'email_on_retry': False,
}

dag, tasks = make_airflow_dag_kubernetized(
    module_name='dagster_airflow_tests.test_project.dagster_airflow_demo',
    pipeline_name='demo_pipeline',
    environment_dict=yaml.load(ENVIRONMENT),
    dag_kwargs={'default_args': DEFAULT_ARGS, 'max_active_runs': 1},
    image='dagster-airflow-demo',
    namespace='default',
)

Diff Detail

Repository
R1 dagster
Branch
k8s_operator
Lint
Lint OK
Unit
No Unit Test Coverage

Event Timeline

natekupp created this revision.Aug 12 2019, 9:44 PM
natekupp edited the summary of this revision. (Show Details)Aug 12 2019, 11:20 PM
natekupp edited the test plan for this revision. (Show Details)
natekupp added a reviewer: Restricted Project.
schrockn requested changes to this revision.Aug 13 2019, 3:23 AM
schrockn added a subscriber: schrockn.

this looks reasonable overall. if python modules were easier to manage for us i might push for a separate module for this but meh for now.

My request would be to go through all the todos and make issues for the ones that make sense.

python_modules/dagster-airflow/dagster_airflow/operators/airflow_kubernetes_pod_operator.py
2 ↗(On Diff #3621)

maybe note what version

python_modules/dagster-airflow/dagster_airflow/operators/kubernetes_operator.py
27

hmmm maybe explain what is going on here

35

what does this argument do?

42

whats this replace all about? also let's stay consistent on ' versus "

68–72

not really clear to me what the actual actions items are from the description

79

let's actually implement this error/warning

192–193

resolve or make them issues

This revision now requires changes to proceed.Aug 13 2019, 3:23 AM
max added a subscriber: max.Aug 13 2019, 3:08 PM
max added inline comments.
python_modules/dagster-airflow/dagster_airflow/operators/airflow_kubernetes_pod_operator.py
2 ↗(On Diff #3621)

yeah, i would note the commit hash and would also like to split this out into a separate dagster_airflow/vendor/ directory. we should also vendor the python and docker operators.

python_modules/dagster-airflow/dagster_airflow/operators/kubernetes_operator.py
27

won't this fail on py2?

128

i'm not sure what's going on here -- why are we providing a special mechanism for these two environment variables?

130

retrieve

python_modules/dagster-airflow/setup.py
61

we should use extras_require

natekupp updated this revision to Diff 3643.Aug 13 2019, 5:37 PM
natekupp marked 12 inline comments as done.
natekupp edited the summary of this revision. (Show Details)

comments

python_modules/dagster-airflow/dagster_airflow/operators/kubernetes_operator.py
27

I just cleaned this up by using kwargs + check, but see https://github.com/PyCQA/pylint/issues/1835 for future reference

35

its hacky, I'm just pulling it out entirely

42

added a comment and fixed all of the " usage throughout this file

68–72

Fixed this

128

yeah this is wonky, I'm pulling it out

192–193

dropping these

schrockn accepted this revision.Aug 13 2019, 5:49 PM

great let's move forward. seems like next critical thing here will be to set up an integration test

python_modules/dagster-airflow/dagster_airflow/factory.py
266

make issue?

python_modules/dagster-airflow/dagster_airflow/operators/kubernetes_operator.py
40

this is quite strange

This revision is now accepted and ready to land.Aug 13 2019, 5:49 PM
natekupp updated this revision to Diff 3659.Aug 13 2019, 7:52 PM
natekupp marked 2 inline comments as done.

up

python_modules/dagster-airflow/dagster_airflow/factory.py
266

yep, done

python_modules/dagster-airflow/dagster_airflow/operators/kubernetes_operator.py
40

yeah agree, I pulled this out

natekupp updated this revision to Diff 3662.Aug 13 2019, 8:25 PM

move import

natekupp updated this revision to Diff 3664.Aug 13 2019, 8:35 PM

ugh coveralls. rebase master