Skip to content

Dag Daemon Task

Note

This example is a replication of an Argo Workflow example in Hera. The upstream example can be found here.

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
from hera.workflows import DAG, Container, Task, Workflow
from hera.workflows.models import Arguments, HTTPGetAction, Inputs, Parameter, Probe

with Workflow(
    api_version="argoproj.io/v1alpha1",
    kind="Workflow",
    generate_name="dag-daemon-task-",
    entrypoint="daemon-example",
) as w:
    with DAG(
        name="daemon-example",
    ) as invocator:
        Task(
            name="influx",
            template="influxdb",
        )
        Task(
            arguments=Arguments(
                parameters=[
                    Parameter(
                        name="cmd",
                        value="curl -XPOST 'http://{{tasks.influx.ip}}:8086/query' --data-urlencode \"q=CREATE DATABASE mydb\"",
                    )
                ],
            ),
            name="init-database",
            template="influxdb-client",
            depends="influx",
        )
        Task(
            arguments=Arguments(
                parameters=[
                    Parameter(
                        name="cmd",
                        value="for i in $(seq 1 20); do curl -XPOST 'http://{{tasks.influx.ip}}:8086/write?db=mydb' -d \"cpu,host=server01,region=uswest load=$i\" ; sleep .5 ; done",
                    )
                ],
            ),
            name="producer-1",
            template="influxdb-client",
            depends="init-database",
        )
        Task(
            arguments=Arguments(
                parameters=[
                    Parameter(
                        name="cmd",
                        value="for i in $(seq 1 20); do curl -XPOST 'http://{{tasks.influx.ip}}:8086/write?db=mydb' -d \"cpu,host=server02,region=uswest load=$((RANDOM % 100))\" ; sleep .5 ; done",
                    )
                ],
            ),
            name="producer-2",
            template="influxdb-client",
            depends="init-database",
        )
        Task(
            arguments=Arguments(
                parameters=[
                    Parameter(
                        name="cmd",
                        value="curl -XPOST 'http://{{tasks.influx.ip}}:8086/write?db=mydb' -d 'cpu,host=server03,region=useast load=15.4'",
                    )
                ],
            ),
            name="producer-3",
            template="influxdb-client",
            depends="init-database",
        )
        Task(
            arguments=Arguments(
                parameters=[
                    Parameter(
                        name="cmd",
                        value='curl --silent -G http://{{tasks.influx.ip}}:8086/query?pretty=true --data-urlencode "db=mydb" --data-urlencode "q=SELECT * FROM cpu"',
                    )
                ],
            ),
            name="consumer",
            template="influxdb-client",
            depends="producer-1 && producer-2 && producer-3",
        )
    Container(
        daemon=True,
        name="influxdb",
        image="influxdb:1.2",
        readiness_probe=Probe(
            http_get=HTTPGetAction(
                path="/ping",
                port=8086,
            ),
            initial_delay_seconds=5,
            timeout_seconds=1,
        ),
    )
    Container(
        inputs=Inputs(
            parameters=[
                Parameter(
                    name="cmd",
                )
            ],
        ),
        name="influxdb-client",
        args=["{{inputs.parameters.cmd}}"],
        command=["sh", "-c"],
        image="appropriate/curl:latest",
    )

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: dag-daemon-task-
spec:
  entrypoint: daemon-example
  templates:
  - name: daemon-example
    dag:
      tasks:
      - name: influx
        template: influxdb
      - name: init-database
        depends: influx
        template: influxdb-client
        arguments:
          parameters:
          - name: cmd
            value: curl -XPOST 'http://{{tasks.influx.ip}}:8086/query' --data-urlencode
              "q=CREATE DATABASE mydb"
      - name: producer-1
        depends: init-database
        template: influxdb-client
        arguments:
          parameters:
          - name: cmd
            value: for i in $(seq 1 20); do curl -XPOST 'http://{{tasks.influx.ip}}:8086/write?db=mydb'
              -d "cpu,host=server01,region=uswest load=$i" ; sleep .5 ; done
      - name: producer-2
        depends: init-database
        template: influxdb-client
        arguments:
          parameters:
          - name: cmd
            value: for i in $(seq 1 20); do curl -XPOST 'http://{{tasks.influx.ip}}:8086/write?db=mydb'
              -d "cpu,host=server02,region=uswest load=$((RANDOM % 100))" ; sleep
              .5 ; done
      - name: producer-3
        depends: init-database
        template: influxdb-client
        arguments:
          parameters:
          - name: cmd
            value: curl -XPOST 'http://{{tasks.influx.ip}}:8086/write?db=mydb' -d
              'cpu,host=server03,region=useast load=15.4'
      - name: consumer
        depends: producer-1 && producer-2 && producer-3
        template: influxdb-client
        arguments:
          parameters:
          - name: cmd
            value: curl --silent -G http://{{tasks.influx.ip}}:8086/query?pretty=true
              --data-urlencode "db=mydb" --data-urlencode "q=SELECT * FROM cpu"
  - name: influxdb
    daemon: true
    container:
      image: influxdb:1.2
      readinessProbe:
        initialDelaySeconds: 5
        timeoutSeconds: 1
        httpGet:
          path: /ping
          port: 8086
  - name: influxdb-client
    container:
      image: appropriate/curl:latest
      args:
      - '{{inputs.parameters.cmd}}'
      command:
      - sh
      - -c
    inputs:
      parameters:
      - name: cmd

Comments