Ability to add timezone and extra user payload to calendar gateway

[VaibhavPage]

No description provided.

[magaldima]

other than one comment, lookds good

[magaldima]

any reason this is a string and not []byte ?

[VaibhavPage]

Most probably user will put JSON string as UserPayload because it is the easiest way to parse it in workflow

[inteledyne]

i can't find any examples of how to pass this userpayload to the workflow that is triggered by the sensor.

[VaibhavPage]

the example would be -

apiVersion: argoproj.io/v1alpha1
kind: Sensor
metadata:
  name: calendar-sensor
  labels:
    sensors.argoproj.io/sensor-controller-instanceid: argo-events
spec:
  template:
    spec:
      containers:
        - name: "sensor"
          image: "argoproj/sensor"
          imagePullPolicy: Always
      serviceAccountName: argo-events-sa
  dependencies:
    - name: "calendar-gateway:interval"
  eventProtocol:
    type: "HTTP"
    http:
      port: "9300"
  triggers:
    - template:
        name: calendar-workflow-trigger
        group: argoproj.io
        version: v1alpha1
        kind: Workflow
        source:
          inline: |
            apiVersion: argoproj.io/v1alpha1
            kind: Workflow
            metadata:
              generateName: calendar-workflow-
            spec:
              entrypoint: whalesay
              arguments:
                parameters:
                - name: message
                  # this is the value that should be overridden
                  value: hello world
              templates:
              - name: whalesay
                inputs:
                  parameters:
                  - name: message
                container:
                  image: docker/whalesay:latest
                  command: [cowsay]
                  args: ["{{inputs.parameters.message}}"]
      resourceParameters:
        - src:
            event: "calendar-gateway:interval"
            path: userPayload
          dest: spec.arguments.parameters.0.value

[inteledyne]

Love it, and is that userPayload set in the gateway? I'm testing a simple POC that involves passing static values to the same workflow for different cron schedules/sensors. Following is what I'm trying to do -

resourceParameters: - src: event: "calendar-gateway:interval" key: "sql_script" value: "[{ 'sqlfile': 'test_1.sql', 'schedulekey': 'recent_records' }]" dest: spec.arguments.parameters.0.value

[VaibhavPage]

That would be in gateway-configmap. e.g calendar gateway configmap would look like -

apiVersion: v1
kind: ConfigMap
metadata:
  name: calendar-gateway-configmap
data:
  interval: |-
    interval: 10s
    userPayload: "{\"hello\": \"world\"}"
  schedule: |-
    schedule: "30 * * * *"
    userPayload: {\"hey\": \"there\"}

To pass json as userPayload, you need to escape that json to get the string.

[inteledyne]

That works perfectly thank you.

[inteledyne]

If I want to have a 1-1 mapping between cron schedule and sensor, what is the best method? Maybe something like this?

apiVersion: v1 kind: ConfigMap metadata: name: calendar-gateway-configmap data: interval-recent-records: |- interval: 60s userPayload: "[{ \"sqlfile\": \"test_1.sql\", \"schedulekey\": \"recent_records\" }]" interval-next-unit-test: interval: 300s userPayload: "[{ \"sqlfile\": \"test_2.sql\", \"schedulekey\": \"recent_records\" }]" schedule: |- schedule: "10 * * * *"

Then in my sensor I have this?

resourceParameters: - src: event: "calendar-gateway:interval-recent-records" path: userPayload dest: spec.arguments.parameters.0.value

[inteledyne]

Is a completely separate configmap required for each unique interval associated with the calendar gateway?

[VaibhavPage]

Gateway always has a single configmap that contains all the configurations.
Do you want to trigger the same workflow for different intervals/schedules or different workflows?

[inteledyne]

Different workflows, different intervals and schedules.

[inteledyne]

ScheduleA - 6pm
ScheduleA - Every 5 minutes
ScheduleB - 9am
Schedule C - 9am, weekdays only

[VaibhavPage]

Cool. Check these example out,

https://github.com/argoproj/argo-events/blob/master/examples/sensors/webhook-http-boolean.yaml
https://github.com/argoproj/argo-events/blob/master/examples/sensors/webhook-http-dependency-groups.yaml

Basically in your case, sensor will have four dependencies ScheduleA, B, C, D. Each dependency will belong to separate group and you would just OR these groups.

Why group dependencies? Because sensor acts as barrier waiting for receive events from one or more gateways. And historically it only supported ANDing the dependencies, so, basically waiting for all events to happen. In your case, you want to trigger workflows, if Schedule A or B or C or D happens, hence you group them and apply OR.

[inteledyne]

I got it working - here's my test -

[VaibhavPage]

So the sensor would look like ,

apiVersion: argoproj.io/v1alpha1
kind: Sensor
metadata:
  name: your-sensor
  labels:
    sensors.argoproj.io/sensor-controller-instanceid: argo-events
spec:
  template:
    spec:
      containers:
        - name: "sensor"
          image: "argoproj/sensor"
          imagePullPolicy: Always
      serviceAccountName: argo-events-sa
  # defines list of all events sensor will accept
  dependencies:
    - name: "calendar-gateway:interval_recent_records"
    - name: "calendar-gateway:interval_max_volume"
    - name: "calendar-gateway:schedule"
  # divides event dependencies into groups
  dependencyGroups:
    - name: "group_1"
      dependencies:
        -  "calendar-gateway:interval_recent_records"
    - name: "group_2"
      dependencies:
        - "calendar-gateway:interval_max_volume"
    - name: "group_3"
      dependencies:
        - "calendar-gateway:schedule"
  # either  "calendar-gateway:interval_recent_records" or "calendar-gateway:interval_max_volume" or  "calendar-gateway:schedule" happens
  circuit: "group_1 || group_2 || group3"
  eventProtocol:
    type: "HTTP"
    http:
      port: "9300"
  triggers:
    - template:
        when:
          all:
            - "group_1"
        name: trigger-1
        group: argoproj.io
        version: v1alpha1
        kind: Workflow
        source:
          inline: |
            apiVersion: argoproj.io/v1alpha1
            kind: Workflow
            metadata:
              generateName: hello-world-
            spec:
              entrypoint: whalesay
              templates:
                - name: whalesay
                  container:
                    args:
                      - "hello world"
                    command:
                      - cowsay
                    image: "docker/whalesay:latest"
      resourceParameters:
        - src:
            event: "calendar-gateway:interval_recent_records"
            path: userPayload
          dest: spec.arguments.parameters.0.value
    - template:
        when:
          all:
            - "group_2"
        name: trigger-2
        group: argoproj.io
        version: v1alpha1
        kind: Workflow
        source:
          inline: |
            apiVersion: argoproj.io/v1alpha1
            kind: Workflow
            metadata:
              generateName: hello-world-2-
            spec:
              entrypoint: whalesay
              templates:
                - name: whalesay
                  container:
                    args:
                      - "hello world"
                    command:
                      - cowsay
                    image: "docker/whalesay:latest"
      resourceParameters:
        - src:
            event:  "calendar-gateway:interval_max_volume"
            path: userPayload
          dest: spec.arguments.parameters.0.value
    - template:
        when:
          all:
            - "group_3"
        name: trigger-3
        group: argoproj.io
        version: v1alpha1
        kind: Workflow
        source:
          inline: |
            apiVersion: argoproj.io/v1alpha1
            kind: Workflow
            metadata:
              generateName: hello-world-2-
            spec:
              entrypoint: whalesay
              templates:
                - name: whalesay
                  container:
                    args:
                      - "hello world"
                    command:
                      - cowsay
                    image: "docker/whalesay:latest"

[VaibhavPage]

The sensor spec looks long because all workflows are inlined. You can actually store workflow definitions on git, s3, file, configmap etc and refer them. https://github.com/argoproj/argo-events/blob/master/docs/trigger-guide.md#how-to-define-a-trigger

[inteledyne]

Interesting, so this is essentially how you would define multiple sensor triggers in the same sensor reacting on multiple events coming from the gateway, firing different workflows.

As an alternative, you could also specify a separate sensor per trigger, each sensor pointing to a separate workflow, correct? I think this approach might make it easier to build out a dynamic workflow generation and deployment wrapper on top of it.

[inteledyne]

Also, I recognize there is probably not the right place to ask this question, but is there a way to run specific steps in a workflow, say in a triage situation for a data pipeline where the pipeline is partially processed?

[VaibhavPage]

yes, you can have a separate sensor that handles a single trigger.

About the workflow question, I would raise that in argo workflow project / slack channel

[janmasarik]

Hey @VaibhavPage , 👋

What is the recommended way how to parse userPayload json in the workflow?

I would like to pass:

apiVersion: v1
kind: ConfigMap
metadata:
  name: calendar-gateway-configmap
data:
  interval-json: |-
    interval: 10s
    userPayload: "{\"domian\": \"argo.io\", \"another-param\": \"bar\"}"

and then, in workflow insert the domain and another-param to 2 different paramters, like this:

apiVersion: argoproj.io/v1alpha1
kind: Sensor
metadata:
  name: calendar-sensor
  labels:
    sensors.argoproj.io/sensor-controller-instanceid: argo-events
spec:
  template:
    spec:
      containers:
        - name: "sensor"
          image: "argoproj/sensor"
          imagePullPolicy: Always
      serviceAccountName: argo-events-sa
  dependencies:
    - name: "calendar-gateway:interval"
  eventProtocol:
    type: "HTTP"
    http:
      port: "9300"
  triggers:
    - template:
        name: calendar-workflow-trigger
        group: argoproj.io
        version: v1alpha1
        kind: Workflow
        source:
          inline: |
            apiVersion: argoproj.io/v1alpha1
            kind: Workflow
            metadata:
              generateName: calendar-workflow-
            spec:
              entrypoint: whalesay
              arguments:
                parameters:
                - name: message
                  # this is the value that should be overridden
                  value: hello world
                - name: another-message
                  value: plz-work
              templates:
              - name: whalesay
                inputs:
                  parameters:
                  - name: message
                  - name: another-message
                container:
                  image: docker/whalesay:latest
                  command: [cowsay]
                  args: ["{{inputs.parameters.message}}", "{{inputs.parameters.another-message}}"]
      resourceParameters:
        - src:
            event: "calendar-gateway:interval"
            path: userPayload.domian
          dest: spec.arguments.parameters.0.value
        - src:
            event: "calendar-gateway:interval"
            path: userPayload.another-param
          dest: spec.arguments.parameters.1.value

Btw, thanks a lot for your awesome work. I really enjoy learning argo-events and seeing how it grows every week! :-)

[VaibhavPage]

As userPayload is a string and not a JSON, you won't be able to access any particular substring using resourceParameters.

I would suggest that you pass the userPayload as the first argument to container command and then convert string to JSON in your process baked into the image.

The source would like following,

        source:
          inline: |
            apiVersion: argoproj.io/v1alpha1
            kind: Workflow
            metadata:
              generateName: calendar-workflow-
            spec:
              entrypoint: whalesay
              arguments:
                parameters:
                - name: message
                  # this is the value that should be overridden
                  value: hello world
              templates:
              - name: whalesay
                inputs:
                  parameters:
                  - name: message
                container:
                  image: <your image>  # It will convert userPayload string to JSON
                  command: [<your command>]
                  args: ["{{inputs.parameters.message}}"]
      resourceParameters:
        - src:
            event: "calendar-gateway:interval"
            path: userPayload
          dest: spec.arguments.parameters.0.value

[janmasarik]

Understood, that's actually what I expected, but still had a small hope that there is some nicer solution for this :-)

Thanks for the answer and your amazing work here!