Introducing On Demand Workflow Triggers with Customizable Conditions

Edge Delta helps customers process their log data as it’s created, where it’s created – before it’s indexed in an observability platform downstream. By doing so, teams can power their dashboards downstream without having to index complete raw datasets. However, there are scenarios, such as deployments and anomalies, when customers want to route raw logs, along with the analytics our platform derives. In these instances, teams can investigate any issues within their observability platform.

To accommodate this need, Edge Delta recently added an event-based workflow that triggers raw log forwarding at the Edge Delta endpoint for a specified duration. Normally, our workflows start to run when the agent has a heartbeat, and last throughout its lifecycle. In this case, you’ll want to run the log forwarding workflow under predefined conditions.

For example, perhaps you want to capture raw logs for a given service 15 minutes after it is deployed.

Our team has a backend service that’s deployed on Kubernetes, and we want to start the workflow for the pods of that specific service when the deployment is done. Normally, we would define the workflow and run the Edge Delta agent forever. The problem with that approach is when the service is not deployed, the workflow will still run and the agent will pull the input sources. This will cause the agent to use resources redundantly.

By specifying an on-demand workflow trigger, we can easily solve this issue. Later in this blog post, we will go over how to start using conditional workflows.

Defining the conditional workflow

To define a conditional workflow, we need to write our workflow-triggering conditions (or events) under workflow_conditions in the agent configuration:

workflow_conditions:
 - name: run_my_workflow
 type: on_demand


Each workflow condition has a name and a type. The name is a unique identifier for each condition, which we will later use to specify the conditions within our workflow. The type is basically the condition type, which can only be on_demand at the time being.

Now we can define the workflow. Assume our pods have "my-deployment-" prefix, and we will use S3 as the output destination.

The full agent configuration is:

version: v2


agent_settings:
 tag: on-demand-test


inputs:
 kubernetes:
 - labels: "deployment-input"
 include:
 - "pod=^my-deployment-.*$,namespace=default"
 exclude: # exlude has higher priority
 - "namespace=edgedelta"


outputs:
 streams:
 - name: my-s3-streamer
 type: s3stream
 aws_key_id: ********
 aws_sec_key: ********
 bucket: log-forward-bucket
 region: us-west-2
 features: log


workflow_conditions:
 - name: run_my_workflow
 type: on_demand


workflows:
 conditional-workflow:
 input_labels:
 - deployment-input
 conditions:
 - run_my_workflow
 destinations:
 - my-s3-streamer


Here we defined the Kubernetes input under inputs.kubernetes with label deployment-input, the output under outputs.streams with name my-s3-streamer, and used the workflow condition we’ve defined before in workflows.conditional-workflow.conditions. We have specified the log output Amazon S3 bucket as log-forward-bucket.

Do not forget to specify log in the features of the output destination. An output destination with log feature is needed for on-demand log forwarding to function.

Deploying the agent

Go to app.edgedelta.com and log in to your account. If you don’t have one, you can sign up and start using for free. After logging in, create a new agent configuration by navigating Agent Settings > Create Configuration. Then, create the agent config by simply using the config snippet above.

After creating the config, we will be able to deploy the agent using the deploy instructions under Actions > Deploy Instructions > Kubernetes:

kubectl create namespace edgedelta
kubectl create secret generic ed-api-key \
 --namespace=edgedelta \
 --from-literal=ed-api-key="<INSERT API KEY HERE>"
kubectl apply -f https://edgedelta.github.io/k8s/edgedelta-agent.yml


Let’s see if the pods are created properly:

kubectl get pods -n edgedelta


Triggering the workflow

Now that the Edge Delta agent is deployed, we can trigger the workflow by simply sending a POST request to /v1/orgs/<org-id>/on_demand endpoint.

You will need to use an API token to send a trigger request. If you don’t have one yet, create one by navigating to app.edgedelta.com/organization#api-tokens and using Create Token. Select the All future and current agent configurations option from Add Permissions > Agent Configurations. Now select the permission type as Write, and add the permission to the token. This is the only permission you will need to use the /on_demand endpoint.

We will use the POST request body to specify the on demand request attributes. The body have a few JSON fields:

• name: This is the workflow condition’s name
• future_duration: Duration of how long the workflow will run. This is in time.Duration format.
• attributes: Custom attributes to filter the on demand request through the agent tag, source type, deployment host, deployment region etc.

The attributes field is an optional object with the type of map[string]string. Attributes are used to filter the on demand request to be processed only in very specific agent deployments. The possible attributes are listed below:

For the source type (src_type), the following source type attributes can be used:

For more information on the endpoint, please refer to the docs.

Consider using the below snippet as the request body to the /on_demand endpoint:

{
 "name": "on-demand-wf-1",
 "future_duration": "1h",
 "attributes": {
 "tag": "agent-prod",
 "src_type": "K8s",
 "k8s_namespace": "prod-v2"
 }
}


This on-demand request will only trigger the conditional workflow named on-demand-wf-1, which has an agent that is tagged as agent-prod and runs in Kubernetes with namespace prod-v2. Also, it will set the future running duration to one hour.

Coming back to our example, we can send a request such as below:

{
 "name": "run_my_workflow",
 "future_duration": "1h",
 "attributes": {
 "tag": "on-demand-test",
 "src_type": "K8s"
 }
}


We can send the request using cURL as below. Don’t forget to replace <YOUR-ORGANIZATION-ID> with your organization ID and $YOUR_ED_API_TOKEN with the API token you have created before. If you don’t know your organization ID, you can find it on app.edgedelta.com/organization.

curl --location --request POST 'https://api.edgedelta.com/v1/orgs/<YOUR-ORGANIZATION-ID>/on_demand' \
 --header "X-ED-API-Token: $YOUR_ED_API_TOKEN" \
 --header 'Content-Type: application/json' \
 --data-raw '{"name":"run_my_workflow","future_duration":"1h","attributes": {"tag":"on-demand-test","src_type":"K8s"}}'


Or, by using httpie:

echo '{"name":"run_my_workflow","future_duration":"1h","attributes":{"tag":"on-demand-test","src_type":"K8s"}}' | https POST api.edgedelta.com/v1/orgs/<YOUR-ORGANIZATION-ID>/on_demand X-ED-API-Token:$YOUR_ED_API_TOKEN


The response will be 201 OK with a payload exactly as the on demand request we have sent:

{
 "name": "run_my_workflow",
 "future_duration": "1h",
 "attributes": {
 "tag": "on-demand-test",
 "src_type": "K8s"
 }
}


Checking the artifacts

Let’s check our bucket’s contents to see if the forwarded logs are present:

$ aws s3 ls s3://log-forward-bucket/2022/10/03/14/on-demand-test/gke-on-demand-cluster-default-pool-2010be2b-6cmw/


2022-10-03 17:55:43  229409 e7f95efb-92fc-4b4e-91e6-d040c5b50203.log.gz


$ aws s3 cp s3://log-forward-bucket/2022/10/03/14/on-demand-test/gke-on-demand-cluster-default-pool-2010be2b-6cmw/e7f95efb-92fc-4b4e-91e6-d040c5b50203.log.gz .
$ gunzip e7f95efb-92fc-4b4e-91e6-d040c5b50203.log.gz
$ head -n 3 e7f95efb-92fc-4b4e-91e6-d040c5b50203.log


{"name":"conditional_workflow_forward","source_type":"K8s","source_name":"default#my-deployment-74db5f4785-w6fh5#flog","log_type":"log","timestamp":1664808939355644843,"message":"[03/10/2022:14:52:12 +0000] \"POST /v1/logout HTTP/1.1\" 200 11738"}
{"name":"conditional_workflow_forward","source_type":"K8s","source_name":"default#my-deployment-74db5f4785-w6fh5#flog","log_type":"log","timestamp":1664808939355646634,"message":"[03/10/2022:14:52:12 +0000] \"PUT /v1/logout HTTP/1.0\" 404 4683"}
{"name":"conditional_workflow_forward","source_type":"K8s","source_name":"default#my-deployment-74db5f4785-w6fh5#flog","log_type":"log","timestamp":1664808939355838579,"message":"[03/10/2022:14:52:12 +0000] \"POST /v1/login HTTP/1.0\" 200 5678 "}


Conclusion

Edge Delta’s on-demand log forwarding allows you to better control the data you index downstream. Now, you can derive analytics at the source and still capture raw data whenever it is beneficial to do so, such as during deployments or anomalies. As a result, you can allocate your observability capacity toward the data that’s most important to your team.