
The Center for Internet Security (CIS) benchmarks for AWS are widely used by cloud teams as guidance to configure their AWS accounts securely. With Flowpipe's new [AWS CIS](https://hub.flowpipe.io/mods/turbot/aws_cis) mod you can detect misconfigurations and apply fixes according to the v3.0.0 or v4.0.0 recommendations. Let's run v4.0.0's section 2 which focuses on storage.

```bash
$ flowpipe pipeline run cis_v400_2
```

Flowpipe walks you through each subsection, finds resources that don't comply with the recommendations, and offers you the choice to skip each resource or apply the recommendation. Here's how that looks in progress. We've arrived at section 2.1.4, and are running a control that checks whether an S3 bucket blocks public access as the recommendation says it should. Flowpipe has found two buckets that don't comply. We opted to skip the first one, and are ready to remediate the second.


![cis run in progess](/images/blog/2024-10-compliance-workflows-for-aws/cis-run-in-progress-1.png)

Here we've chosen `Block public access`. Flowpipe takes the action, reports the outcome, and proceeds. 

![aws console before fix](/images/blog/2024-10-compliance-workflows-for-aws/cis-run-in-progress-2.png)

You can step through all of section 2 in this way. You could even step through the entire CIS benchmark, section by section and control by control. 

## Under the hood: AWS Compliance

The [AWS CIS](https://hub.flowpipe.io/mods/turbot/aws_cis) mod is actually a thin wrapper around another new mod, [AWS Compliance](https://hub.flowpipe.io/mods/turbot/aws_compliance), which provides a suite of pipelines that do the work of detecting misconfigurations and enabling human-supervised correction. `AWS Compliance` in turn relies on two helpers: Flowpipe's [library mod for AWS](https://hub.flowpipe.io/mods/turbot/aws) and a [helper mod](https://github.com/turbot/flowpipe-mod-detect-correct) that implements the detect-and-correct pattern.

Here's how things look from the perspective of the installed `AWS CIS` mod.

```bash
~/flowpipe-mod-aws-cis$ flowpipe mod list
aws_cis
└── github.com/turbot/flowpipe-mod-aws-compliance@v1.0.0
    ├── github.com/turbot/flowpipe-mod-aws@v1.0.0
    └── github.com/turbot/flowpipe-mod-detect-correct@v1.0.0
```

You can list the top-level CIS pipelines like this.

```bash
~/flowpipe-mod-aws-cis$ flowpipe pipeline list | grep aws_cis
aws_cis           cis_v300            The CIS Amazon Web Services Foundations Benchmark provides...
aws_cis           cis_v300_1
aws_cis           cis_v300_1_1
aws_cis           cis_v300_1_10
```

You can also list the supporting detect-and-correct pipelines like this.

```bash
~/flowpipe-mod-aws-cis$ flowpipe pipeline list | grep detect_and_correct
aws_compliance    detect_and_correct_accounts_without_alternate_contact_security_registered                             Detects account alternate ...
aws_compliance    detect_and_correct_apigateway_rest_api_stage_if_xray_tracing_disabled                                 Detects unattached EIPs ...
...
```

One of those, `detect_and_correct_s3_buckets_if_publicly_accessible`, is the pipeline that's used by CIS 2.1.4. You can also install `AWS Compliance`  separately and use that pipeline directly in a standalone context.

```bash
~/flowpipe-mod-aws-compliance$ flowpipe pipeline run detect_and_correct_s3_buckets_with_block_public_access_disabled
```

By default this works just as we've already seen: you're prompted with each bucket that doesn't block access, and in each case you choose to skip or apply the action. But making all those decisions can get old. What if you want to apply the fix to all non-compliant buckets? You can do that by setting a mod variable, `approvers`, to the empty list. Flowpipe will then automatically apply the fix across all your buckets.

Now suppose you want to apply the same rule to new buckets as they're created? To do that, configure the pipeline to run on a schedule. Again it's just a matter of setting variables. To activate a scheduled run of the pipeline, set the variable `s3_bucket_access_enabled_trigger_enabled` to `true`. And if you want to run on a schedule other than the default, which is every 15 minutes, also set `s3_bucket_access_enabled_trigger_schedule` to use either a [named interval](https://flowpipe.io/docs/flowpipe-hcl/trigger/schedule#arguments) or a custom cron expression.

## Degrees of freedom: benchmarks, controls, and pipelines

  If your priority is CIS, we've got you covered with the [AWS CIS](https://hub.flowpipe.io/mods/turbot/aws_cis) mod. It provides the flexibility to run the whole suite, or just one section, or even individual controls, with interactive supervision on a per-resource basis. If you need even more flexibility, the supporting [AWS Compliance](https://hub.flowpipe.io/mods/turbot/aws_compliance) pipelines are available to be used in other ways, with or without human approval, on a one-time or scheduled basis. But there's even more flexibility than that. If you need to act on AWS resources in ways other than `AWS Compliance` does, the [AWS](https://hub.flowpipe.io/mods/turbot/aws) library mod gives you the raw ingredients to run different kinds of queries and take different kinds of actions. Use these mods in whatever ways make sense to you, and [let us know](https://turbot.com/community/join) how it goes.


## See it in action

<div className="flex justify-center">

<iframe
    class="youtube-video"
    src="https://www.youtube-nocookie.com/embed/I3RtrxQM6T4" 
    frameBorder="0"
    allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture; web-share"
    allowFullScreen
    title="Remediate AWS CIS v3.0 and v4.0 with Flowpipe's interactive wizard"
>
</iframe>
</div>


Flowpipe's new AWS CIS mod finds misconfigurations and applies fixes, with step-by-step approval.

Remediate AWS CIS v3.0 and v4.0 with Flowpipe's interactive wizard


  Flowpipe supports workflows that require decisions based on human input. As detailed in our blog posts [Workflow for DevOps: Approvals and inputs to engage your team](https://flowpipe.io/blog/human-interaction) and [Control AWS costs with 29 FinOps workflows](https://flowpipe.io/blog/finops-workflows-for-aws), pipelines can push that decision-making to Slack, MSTeams, or email. Now, these interactions can happen directly in your terminal too.

  ## Decisions in your terminal

Here's an example: a pipeline called `check_old_branches` that asks whether to delete stale branches in a repository.

```
$ flowpipe pipeline run check_old_branches
┃ Do you want to delete branch add-tables?
┃ > Skip
┃   Delete
```

Flowpipe pauses for a decision. Let's skip this one, and proceed.

```
┃ Do you want to delete branch update-slack-community-link?
┃   Skip
┃ > Delete
```

Here we'll choose Delete.

If those are the only two branches matching a Steampipe query that finds stale branches, Flowpipe concludes with this report.

```
Skipped branch add-tables
Deleted branch update-slack-community-link
```

## How it works

To look for branches older than 30 days that also lack pull requests, the pipeline runs a [query step](https://flowpipe.io/docs/flowpipe-hcl/step/query) that joins two Steampipe tables: [github_branch](https://hub.steampipe.io/plugins/turbot/github/tables/github_branch) and [github_pull_request](https://hub.steampipe.io/plugins/turbot/github/tables/github_pull_request).

Here's the pipeline that `check_old_branches` calls to make a decision for each branch. It receives a branch name from the query step, prompts to `Skip` or `Delete`, then calls the `skip_branch` or `delete_branch` pipelines accordingly.

```hcl
pipeline "handle_one_branch" {
  param "name" {
    type        = string
    description = "The name of the branch."
  }

  step "input" "approve_deletion" {
    type     = "button"
    prompt   = "Do you want to delete branch ${param.name}?"
    notifier = notifier.default
    option "Skip" {}
    option "Delete" {}
  }

  step "pipeline" "delete_branch" {
    pipeline = pipeline.delete_branch
    args = {
      name = param.name
    }
    if = step.input.approve_deletion.value == "Delete"
  }

  step "pipeline" "skip_branch" {
    pipeline = pipeline.skip_branch
    args = {
      name = param.name
    }
    if = step.input.approve_deletion.value == "Skip"
  }
}
```

To run a pipeline like this, you'd first install the [GitHub mod for Flowpipe](https://hub.flowpipe.io/mods/turbot/github) to gain access to the [delete_branch](https://hub.flowpipe.io/mods/turbot/github/pipelines/github.pipeline.delete_branch) pipeline.

## Develop in the terminal, then interact with your team

The example uses the default [notifier](https://flowpipe.io/docs/reference/config-files/notifier) to interact in the terminal, versus alternate notifiers that you might configure for [Slack](https://flowpipe.io/docs/reference/config-files/integration/slack), [MSTeams](https://flowpipe.io/docs/reference/config-files/integration/msteams), or [email](https://flowpipe.io/docs/reference/config-files/integration/email). If you're the decision-maker, you can handle approval-based workflows directly. When other team members need to approve, you can route decisions to them using their preferred communication channels. 

And of course you can tap into two powerful ecosystems: Steampipe's [suite of API plugins](https://hub.steampipe.io), and Flowpipe's [collection of utility pipelines](https://hub.flowpipe.io).

Give this new feature a try, and [let us know](https://turbot.com/community/join) how it enhances your workflow automation!


Flowpipe's input step now works in your terminal, as well as via Slack, MSTeams, and email.

Interactive DevOps workflows in your terminal


In the world of cloud infrastructure management, consistent and accurate resource tagging is crucial. It's nice to be able to detect when tags deviate from your standards, but even better to correct those deviations automatically. Enter the [AWS Tags mod for Flowpipe](https://github.com/turbot/flowpipe-mod-aws-tags), a new tool that enables you to automate tag management using a declarative ruleset. Let's take a look.

## Add mandatory tags

Here's a Steampipe query that reveals incorrectly-tagged S3 buckets.

```sql
select name, tags from aws_s3_bucket where name ~ 'flowpipe'
+--------------------+------------------+
| name               | tags             |
+--------------------+------------------+
| flowpipe-bucket-01 | {"cc":"default"} |
| flowpipe-bucket-02 | <null>           |
+--------------------+------------------+
```

First, we'd like to apply the mandatory tag `env:dev` to all of them. To do that, we'll use the `detect_and_correct_s3_buckets_with_incorrect_tags` pipeline in the AWS Tags mod.

Per the README, the setup entails:

- Cloning the mod

- Visiting its directory and installing dependencies (`flowpipe mod install`)

- Starting Steampipe as a service (`steampipe service start`)

- Using `credential_import` to give Flowpipe access to the credentials in Steampipe `aws.spc` 

- Configuring tag rules

We'll use these definitions in `add-notify.fpvars`.

```hcl
approvers = []
base_tag_rules = {
  add = {
    env = "dev"
  }
}
```

With `approvers = []`, the mod will report tag violations to the default notifier, the terminal, but take no action. The only rule is to add our desired mandatory tag.

Now we'll run the pipeline to detect and correct S3 bucket tags, and point it at that package of variables.

```
$ flowpipe pipeline run detect_and_correct_s3_buckets_with_incorrect_tags --var-file=add-notify.fpvars
```

Flowpipe reports the changes it would make.

```
Detected flowpipe-bucket-01 (070715377127/us-east-1/arn:aws:s3:::flowpipe-bucket-01) with incorrect tags. Tags that will be added or updated: env=dev.

Detected flowpipe-bucket-02 (070715377127/us-east-1/arn:aws:s3:::flowpipe-bucket-02) with incorrect tags. Tags that will be added or updated: env=dev.
```

To make the changes, we'll use `add-wizard.fpvars`, with the same tag rules but with `approvers = ["default"]`. That activates an [input step](https://flowpipe.io/docs/flowpipe-hcl/step/input) that will (in this case) prompt to `Skip` or `Apply` the proposed change.

Flowpipe pauses on the first detected violation.

```
$ flowpipe pipeline run detect_and_correct_s3_buckets_with_incorrect_tags --var-file=add-wizard.fpvars
┃ Detected flowpipe-bucket-02 (070715377127/us-east-1/arn:aws:s3:::flowpipe-bucket-02) with incorrect tags. Tags that will be added or updated: env=dev.
┃ > Skip
┃   Apply
```

We'll down-arrow and choose `Apply`. Flowpipe reports:

```
Applied changes to tags on flowpipe-bucket-02.
```

Because we specified the default approver, this interaction happens in the terminal. You could shift the interaction to Slack, MSTeams, or email by naming — in `approvers` — one or more [notifiers](https://flowpipe.io/docs/reference/config-files/notifier) configured with alternate [integrations](https://flowpipe.io/docs/reference/config-files/integration).

We'll choose `Apply` for the second bucket and recheck the situation.

```
+--------------------+------------------------------------------+
| name               | tags                                     |
+--------------------+------------------------------------------+
| flowpipe-bucket-01 | {"cc":"default","env":"dev"}             |
| flowpipe-bucket-02 | {"cc":"default"}                         |
+--------------------+------------------------------------------+
```

That's a good start, but we now want to correct `cc`, that's a bogus tag that some people use for `cost_center`. Let's fix that.


## Update tag keys

We'll use this ruleset in `update-fix.fpvars`.

```hcl
approvers = []

base_tag_rules = {
  update_keys = {
    cost_center = ["~*:^cc|cost_centre$"]
  }
}

incorrect_tags_default_action = "apply"
```

With `approvers = []` the rules run without pausing for input. 

The `update_keys` rule uses pattern matching to converge a set of bogus tags (`cc` or `cost_centre`) onto `cost_center`. 

The default `incorrect_tags_default_action` is `notify`, which would be a dry run, but we want to apply the fix so we set it to `apply`.

We run the pipeline again, with this new package of variables.

```
$ flowpipe pipeline run detect_and_correct_s3_buckets_with_incorrect_tags --var-file=update-fix.fpvars
```

Flowpipe reports:

```
Applied changes to tags on flowpipe-bucket-01.
```

And here's the new situation.

```
+--------------------+---------------------------------------+
| name               | tags                                  |
+--------------------+---------------------------------------+
| flowpipe-bucket-01 | {"cost_center":"default","env":"dev"} |
| flowpipe-bucket-02 | {"env":"dev"}                         |
+--------------------+---------------------------------------+
```

## Tag operations

We've seen `add` and `update_keys`, here's the full set of operations documented in the [README](https://github.com/turbot/flowpipe-mod-aws-tags).

- **Add**. Ensure resources have mandatory tags.

- **Remove**. Remove specified tags.

- **Remove Except**. Inverse of remove: remove all but specified tags.

- **Update Keys**. Rewrite tag keys.

- **Update Values**. Rewrite tag values.

To match tags you can use literal strings, or Postgres regular expressions.

## Automate the fixes

It would be tedious to apply these kinds of changes with one-at-a-time approval. Let's automate with a new ruleset in `continuous-compliance.fpvars`.

```hcl
approvers = []

base_tag_rules = {
  update_keys = {
    cost_center = ["~*:^cc|cost_centre$"]
  }
  add = {
    env = "dev"
  }
}

s3_buckets_with_incorrect_tags_trigger_enabled   = true
s3_buckets_with_incorrect_tags_trigger_schedule  = "daily"
incorrect_tags_default_action                    = "apply"
```

Note that you can combine operations like `add` and `update_keys` to build up a complex ruleset. 

Here we're also using config variables to active the pipeline's [query trigger](https://flowpipe.io/docs/flowpipe-hcl/trigger/query), define its schedule, and define the default action (`apply`). With these settings, Flowpipe will run the pipeline daily and apply the changes we've seen to new buckets.

To run the pipeline on a schedule, use Flowpipe in [server](https://flowpipe.io/docs/run/server) mode.

```
flowpipe server --var-file=continuous-compliance.fpvars
```

Nothing will happen the first day because we've already fixed the buckets. But if we create `flowpipe-bucket-03` with no tags, it will — within a day — automagically be tagged with `env:dev`.

## Continuous compliance for tags

The AWS Tags mod for Flowpipe enables flexible and powerful management of AWS resource tags at scale. With declarative rulesets and automated workflows, you can ensure consistent tagging across your AWS infrastructure with or without manual intervention. Whether you're looking to standardize existing tags, enforce new tagging policies, or maintain ongoing compliance, this mod provides the tools you need to streamline your tagging discipline. Give it a try and please [let us know](https://turbot.com/community/join) how it goes.

## See it in action

<div className="flex justify-center">

<iframe
    class="youtube-video"
    src="https://www.youtube-nocookie.com/embed/FNJt-a_nizY" 
    frameBorder="0"
    allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture; web-share"
    allowFullScreen
    title="Define and enforce tagging standards in AWS"
>
</iframe>
</div>




Use declarative rules to add mandatory tags, remove prohibited tags, and fix misspellings.

Define and enforce tagging standards in AWS


In the world of cloud infrastructure management, consistent and accurate resource tagging is crucial. It's nice to be able to detect when tags deviate from your standards, but even better to correct those deviations automatically. Enter the [Azure Tags mod for Flowpipe](https://github.com/turbot/flowpipe-mod-azure-tags), a new tool that enables you to automate tag management using a declarative ruleset. Let's take a look.

## Add mandatory tags

Here's a Steampipe query that reveals incorrectly-tagged Azure storage accounts.

```sql
select name, tags from azure_storage_account where name ~ 'flowpipe'
+------------------+------------------+
| name             | tags             |
+------------------+------------------+
| flowpipestorage1 | {"cc":"default"} |
| flowpipestorage2 | {}               |
+------------------+------------------+
```

First, we'd like to apply the mandatory tag `env:dev` to all of them. To do that, we'll use the `detect_and_correct_storage_accounts_with_incorrect_tags` pipeline in the Azure Tags mod.

Per the README, the setup entails:

- Cloning the mod

- Visiting its directory and installing dependencies (`flowpipe mod install`)

- Starting Steampipe as a service (`steampipe service start`)

- Using `credential_import` to give Flowpipe access to the credentials in Steampipe `azure.spc` 

- Configuring tag rules

We'll use these definitions in `add-notify.fpvars`.

```hcl
approvers = []
base_tag_rules = {
  add = {
    env = "dev"
  }
}
```

With `approvers = []`, the mod will report tag violations to the default notifier, the terminal, but take no action. The only rule is to add our desired mandatory tag.

Now we'll run the pipeline to detect and correct tags on storage accounts, and point it at that package of variables.

```
$ flowpipe pipeline run detect_and_correct_storage_accounts_with_incorrect_tags --var-file=add-notify.fpvars
```

Flowpipe reports the changes it would make.

```
Detected flowpipestorage1 (288007e6-582c-478d-97cf-c27eea910d65/eastus) with incorrect tags. Existing tags: cc=default New tags: cc=default, env=dev

Detected flowpipestorage2 (288007e6-582c-478d-97cf-c27eea910d65/eastus) with incorrect tags. New tags: env=dev
```

To make the changes, we'll use `add-wizard.fpvars`, with the same tag rules but with `approvers = ["default"]`. That activates an [input step](https://flowpipe.io/docs/flowpipe-hcl/step/input) that will (in this case) prompt to `Skip` or `Apply` the proposed change.

Flowpipe pauses on the first detected violation.

```
$ flowpipe pipeline run detect_and_correct_storage_accounts_with_incorrect_tags --var-file=add-wizard.fpvars
┃ Detected flowpipestorage1 (288007e6-582c-478d-97cf-c27eea910d65/eastus) with incorrect tags. Tags that will be added or updated: env=dev.
┃ > Skip
┃   Apply
```

We'll down-arrow and choose `Apply`. Flowpipe reports:

```
Applied changes to tags on flowpipestorage1.
```

Because we specified the default approver, this interaction happens in the terminal. You could shift the interaction to Slack, MSTeams, or email by naming — in `approvers` — one or more [notifiers](https://flowpipe.io/docs/reference/config-files/notifier) configured with alternate [integrations](https://flowpipe.io/docs/reference/config-files/integration).

We'll choose `Apply` for the second storage account and recheck the situation.

```sql
+--------------------+------------------------------------------+
| name               | tags                                     |
+--------------------+------------------------------------------+
| flowpipestorage1   | {"cc":"default","env":"dev"}             |
| flowpipestorage2   | {"env":"dev"}                            |
+--------------------+------------------------------------------+
```

That's a good start, but we now want to correct `cc`, that's a bogus tag that some people use for `cost_center`. Let's fix that.


## Update tag keys

We'll use this ruleset in `update-fix.fpvars`.

```hcl
approvers = []

base_tag_rules = {
  update_keys = {
    cost_center = ["~*:^cc|cost_centre$"]
  }
}

incorrect_tags_default_action = "apply"
```

With `approvers = []` the rules run without pausing for input. 

The `update_keys` rule uses pattern matching to converge a set of bogus tags (`cc` or `cost_centre`) onto `cost_center`. 

The default `incorrect_tags_default_action` is `notify`, which would be a dry run, but we want to apply the fix so we set it to `apply`.

We run the pipeline again, with this new package of variables.

```
$ flowpipe pipeline run detect_and_correct_storage_accounts_with_incorrect_tags --var-file=update-fix.fpvars
```

Flowpipe reports:

```
Applied changes to tags on flowpipestorage1.
```

And here's the new situation.

```
+--------------------+---------------------------------------+
| name               | tags                                  |
+--------------------+---------------------------------------+
| flowpipestorage1   | {"cost_center":"default","env":"dev"} |
| flowpipestorage2   | {"env":"dev"}                         |
+--------------------+---------------------------------------+
```

## Tag operations

We've seen `add` and `update_keys`, here's the full set of operations documented in the [README](https://github.com/turbot/flowpipe-mod-azure-tags).

- **Add**. Ensure resources have mandatory tags.

- **Remove**. Remove specified tags.

- **Remove Except**. Inverse of remove: remove all but specified tags.

- **Update Keys**. Rewrite tag keys.

- **Update Values**. Rewrite tag values.

To match tags you can use literal strings, or Postgres regular expressions.

## Automate the fixes

It would be tedious to apply these kinds of changes with one-at-a-time approval. Let's automate with a new ruleset in `continuous-compliance.fpvars`.

```hcl
approvers = []

base_tag_rules = {
  update_keys = {
    cost_center = ["~*:^cc|cost_centre$"]
  }
  add = {
    env = "dev"
  }
}

storage_accounts_with_incorrect_tags_trigger_enabled   = true
storage_accounts_with_incorrect_tags_trigger_schedule  = "daily"
incorrect_tags_default_action                          = "apply"
```

Note that you can combine operations like `add` and `update_keys` to build up a complex ruleset. 

Here we're also using config variables to active the pipeline's [query trigger](https://flowpipe.io/docs/flowpipe-hcl/trigger/query), define its schedule, and define the default action (`apply`). With these settings, Flowpipe will run the pipeline daily and apply the changes we've seen to new storage accounts.

To run the pipeline on a schedule, use Flowpipe in [server](https://flowpipe.io/docs/run/server) mode.

```
flowpipe server --var-file=continuous-compliance.fpvars
```

Nothing will happen the first time because we've already fixed the storage accounts. But if we create `flowpipestorage3` with no tags, it will — within a day — automagically be tagged with `env:dev`.

## Continuous compliance for tags

The Azure Tags mod for Flowpipe enables flexible and powerful management of Azure resource tags at scale. With declarative rulesets and automated workflows, you can ensure consistent tagging across your Azure infrastructure with or without manual intervention. Whether you're looking to standardize existing tags, enforce new tagging policies, or maintain ongoing compliance, this mod provides the tools you need to streamline your tagging discipline. Give it a try and please [let us know](https://turbot.com/community/join) how it goes.

## See it in action

<div className="flex justify-center">

<iframe
    class="youtube-video"
    src="https://www.youtube-nocookie.com/embed/ExoZNJPSHks" 
    frameBorder="0"
    allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture; web-share"
    allowFullScreen
    title="Define and enforce tagging standards in Azure"
>
</iframe>
</div>





Define and enforce tagging standards in Azure


In the world of cloud infrastructure management, consistent and accurate resource tagging is crucial. It's nice to be able to detect when GCP labels deviate from your standards, but even better to correct those deviations automatically. Enter the [GCP Labels mod for Flowpipe](https://github.com/turbot/flowpipe-mod-gcp-labels), a new tool that enables you to automate label management using a declarative ruleset. Let's take a look.

## Add mandatory labels

Here's a Steampipe query that reveals incorrectly-labeled GCP buckets.

```sql
select name, labels from gcp_storage_bucket where name ~ 'flowpipe'
+--------------------+------------------+
| name               | labels           |
+--------------------+------------------+
| flowpipe-bucket-01 | {"cc":"default"} |
| flowpipe-bucket-02 | <null>           |
+--------------------+------------------+
```

First, we'd like to apply the mandatory label `env:dev` to all of them. To do that, we'll use the `detect_and_correct_storage_buckets_with_incorrect_labels` pipeline in the GCP Labels mod.

Per the README, the setup entails:

- Cloning the mod

- Visiting its directory and installing dependencies (`flowpipe mod install`)

- Starting Steampipe as a service (`steampipe service start`)

- Using `credential_import` to give Flowpipe access to the credentials in Steampipe `gcp.spc` 

- Configuring label rules

We'll use these definitions in `add-notify.fpvars`.

```hcl
approvers = []
base_label_rules = {
  add = {
    env = "dev"
  }
}
```

With `approvers = []`, the mod will report label violations to the default notifier, the terminal, but take no action. The only rule is to add our desired mandatory label.

Now we'll run the pipeline to detect and correct bucket labels, and point it at that package of variables.

```
$ flowpipe pipeline run detect_and_correct_storage_buckets_with_incorrect_labels --var-file=add-notify.fpvars
```

Flowpipe reports the changes it would make.

```
Detected flowpipe-bucket-01 (flowpipe-bucket-01/encoded-blend-333001/) with incorrect labels. Labels that will be added or updated: env=dev.

Detected flowpipe-bucket-02 (flowpipe-bucket-02/encoded-blend-333001/) with incorrect labels. Labels that will be added or updated: env=dev.
```

To make the changes, we'll use `add-wizard.fpvars`, with the same label rules but with `approvers = ["default"]`. That activates an [input step](https://flowpipe.io/docs/flowpipe-hcl/step/input) that will (in this case) prompt to `Skip` or `Apply` the proposed change.

Flowpipe pauses on the first detected violation.

```
$ flowpipe pipeline run detect_and_correct_storage_buckets_with_incorrect_labels --var-file=add-wizard.fpvars
┃ Detected flowpipe-bucket-01 (flowpipe-bucket-01/encoded-blend-333001/) with incorrect labels. Labels that will be added or updated: env=dev.
┃ > Skip
┃   Apply
```

We'll down-arrow and choose `Apply`. Flowpipe reports:

```
Applied changes to labels on flowpipe-bucket-01.
```

Because we specified the default approver, this interaction happens in the console. You could shift the interaction to Slack, MSTeams, or email by naming — in `approvers` — one or more [notifiers](https://flowpipe.io/docs/reference/config-files/notifier) configured with alternate [integrations](https://flowpipe.io/docs/reference/config-files/integration).

We'll choose `Apply` for the second bucket and recheck the situation.

```
+--------------------+------------------------------------------+
| name               | labels                                   |
+--------------------+------------------------------------------+
| flowpipe-bucket-01 | {"cc":"default","env":"dev"}             |
| flowpipe-bucket-02 | {"env":"dev"}                            |
+--------------------+------------------------------------------+
```

That's a good start, but we now want to correct `cc`, that's a bogus label that some people use for `cost_center`. Let's fix that.


## Update label keys

We'll use this ruleset in `update-fix.fpvars`.

```hcl
approvers = []

base_label_rules = {
  update_keys = {
    cost_center = ["~*:^cc|cost_centre$"]
  }
}

incorrect_labels_default_action = "apply"
```

With `approvers = []` the rules run without pausing for input. 

The `update_keys` rule uses pattern matching to converge a set of bogus labels (`cc` or `cost_centre`) onto `cost_center`. 

The default `incorrect_labels_default_action` is `notify`, which would be a dry run, but we want to apply the fix so we set it to `apply`.

We run the pipeline again, with this new package of variables.

```
$ flowpipe pipeline run detect_and_correct_storage_buckets_with_incorrect_labels --var-file=update-fix.fpvars
```

Flowpipe reports:

```
Applied changes to labels on flowpipe-bucket-01.
```

And here's the new situation.

```
+--------------------+---------------------------------------+
| name               | labels                                |
+--------------------+---------------------------------------+
| flowpipe-bucket-01 | {"cost_center":"default","env":"dev"} |
| flowpipe-bucket-02 | {"env":"dev"}                         |
+--------------------+---------------------------------------+
```

## Label operations

We've seen `add` and `update_keys`, here's the full set of operations documented in the [README](https://github.com/turbot/flowpipe-mod-gcp-labels).

- **Add**. Ensure resources have mandatory labels.

- **Remove**. Remove specified labels.

- **Remove Except**. Inverse of remove: remove all but specified labels.

- **Update Keys**. Rewrite label keys.

- **Update Values**. Rewrite label values.

To match labels you can use literal strings, or Postgres regular expressions.

## Automate the fixes

It would be tedious to apply these kinds of changes with one-at-a-time approval. Let's automate with a new ruleset in `continuous-compliance.fpvars`.

```hcl
approvers = []

base_label_rules = {
  update_keys = {
    cost_center = ["~*:^cc|cost_centre$"]
  }
  add = {
    env = "dev"
  }
}

storage_buckets_with_incorrect_labels_trigger_enabled  = true
storage_buckets_with_incorrect_labels_trigger_schedule = "daily"
incorrect_labels_default_action                        = "apply"
```

Note that you can combine operations like `add` and `update_keys` to build up a complex ruleset. 

Here we're also using config variables to active the pipeline's [query trigger](https://flowpipe.io/docs/flowpipe-hcl/trigger/query), define its schedule, and define the default action (`apply`). With these settings, Flowpipe will run the pipeline daily apply the changes we've seen to new buckets.

To run the pipeline on a schedule, use Flowpipe in [server](https://flowpipe.io/docs/run/server) mode.

```
flowpipe server --var-file=continuous-compliance.fpvars
```

Nothing will happen the first day because we've already fixed the buckets. But if we create `flowpipe-bucket-03` with no labels it will — within a day — automagically be labeled with `env:dev`.

## Continuous compliance for labels

The GCP Labels mod for Flowpipe enables flexible and powerful management of GCP labels at scale. With declarative rulesets and automated workflows, you can ensure consistent labeling across your GCP infrastructure with or without manual intervention. Whether you're looking to standardize existing labels, enforce new labeling policies, or maintain ongoing compliance, this mod provides the tools you need to streamline your labeling discipline. Give it a try and please [let us know](https://turbot.com/community/join) how it goes.

## See it in action

<div className="flex justify-center">

<iframe
    class="youtube-video"
    src="https://www.youtube-nocookie.com/embed/8o11poEgf8c" 
    frameBorder="0"
    allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture; web-share"
    allowFullScreen
    title="Define and enforce labeling standards in GCP"
>
</iframe>
</div>



Use declarative rules to add mandatory labels, remove prohibited labels, and fix misspellings.

Define and enforce labeling standards in GCP


Cloud infrastructure managers are always looking for ways to save costs. It's good to detect underused or unused resources, and even better to automatically streamline or remove them. Enter the [Azure Thrifty mod](https://github.com/turbot/flowpipe-mod-azure-thrifty) for Flowpipe, a new tool that enables you automate cost optimization. Let's see how it works.

## Detect and release unused external IPs

Here's a Steampipe query that finds unattached external IPs. 

```sql
> select name, region from azure_public_ip;
+---------+--------+
| name    | region |
+---------+--------+
| ipaddr1 | eastus |
| ipaddr2 | eastus |
+---------+--------+
```

We're paying for these, and if we don't need them we should release them. To do that, we'll run the `detect_and_correct_compute_addresses_if_unattached` pipeline in the Azure Thrifty mod.

Per the [README](https://github.com/turbot/flowpipe-mod-azure-thrifty?tab=readme-ov-file#azure-thrifty-mod-for-flowpipe), the setup entails:

- Cloning the mod

- Visiting its directory and installing dependencies (`flowpipe mod install`)

- Starting Steampipe as a service (`steampipe service start`)

- Using `credential_import` to give Flowpipe access to the credentials in Steampipe `azure.spc` 

- Optionally, configuring variables that govern the mod's behavior.

Let's run the pipeline.

```
$ flowpipe pipeline run detect_and_correct_compute_addresses_if_unattached --arg='approvers=[""]

Detected unattached Compute address ip-addr-1.
Detected unattached Compute address ip-addr-2.
```

Now let's take action on them. 

## Wizard mode for interactive decisions

In wizard mode, Flowpipe prompts you to decide whether to release each address. It's the default, so we can run the pipeline with no arguments.


```
flowpipe pipeline run detect_and_correct_compute_addresses_if_unattached
```

Now, Flowpipe pauses on each detected address, so you can decide whether to release it. We'll choose `Skip` and keep this one for now.

```
┃ Detected unattached Compute address ip-addr-1.
┃ > Skip
┃   Delete Compute Address
```

Note that this interaction can instead happen in Slack, MSTeams, or email by naming — in `approvers` — one or more notifiers configured with alternate [integrations](https://flowpipe.io/docs/reference/config-files/integration).

Flowpipe proceeds to the next detected address. This time we'll choose `Delete Compute Address`; Flowpipe takes that action and reports the outcome.

```
Deleted Compute address ip-addr-2.
```

Nice! 

## Delete old compute snapshots automatically

You won't want to automatically release IP addresses without checking each one, but you might want to automatically delete compute snapshots older than a threshold number of days. There's another pipeline for that, `detect_and_correct_compute_snapshots_exceeding_max_age`. 

To use that pipeline automatically, you'll activate a [query trigger](https://flowpipe.io/docs/flowpipe-hcl/trigger/query) to query for old snapshots. And you'll set some configuration variables to control how it works. We'll put them into a file, `continuous-compliance.fpvars`.

```hcl
approvers = []
compute_snapshots_exceeding_max_age_trigger_enabled  = true
compute_snapshots_exceeding_max_age_trigger_schedule = "weekly"
compute_snapshots_exceeding_max_age_default_action   = "delete_snapshot"
compute_snapshots_exceeding_max_age_days             = 90
```

Then we'll run Flowpipe in server mode and point it at that package of variables.

```
flowpipe server --var-file=continuous-compliance.fpvars
```

With `approvers = []` the pipeline handles each detected old snapshot without requiring approval. The default action is normally `notify` but here we override to `delete_snapshot`. Now, snapshots older than 90 days will be automatically deleted on a weekly basis.

## Configuration-driven workflow

A detect-and-correct mod like this one doesn't require you to write any SQL or HCL. It's a tool that you configure and then use to perform a range of tasks you'd like to automate, with or without approval. Give Azure Thrifty a try, and please [let us know](https://turbot.com/community/join) how it goes.

## See it in action

<div className="flex justify-center">
<iframe
    class="youtube-video"
    src="https://www.youtube-nocookie.com/embed/lTpHqQK02C0" 
    frameBorder="0"
    allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture; web-share"
    allowFullScreen
    title="Control Azure costs with 24 workflows for FinOps"
>
</iframe>
</div>




FinOps workflows to discover, approve, and automate Azure cost savings recommendations.

Control Azure costs with 24 workflows for FinOps


Cloud infrastructure managers are always looking for ways to save costs. It's good to detect underused or unused resources, and even better to automatically streamline or remove them. Enter the [GCP Thrifty mod](https://github.com/turbot/flowpipe-mod-gcp-thrifty) for Flowpipe, a new tool that enables you automate cost optimization. Let's see how it works.

## Detect and release unused external IPs

Here's a Steampipe query that finds unattached external IPs. 

```sql
> select name, location from gcp_compute_address
+-----------+-------------+
| name      | location    |
+-----------+-------------+
| ip-addr-2 | us-central1 |
| ip-addr-1 | us-central1 |
+-----------+-------------+
```

We're paying for these, and if we don't need them we should release them. To do that, we'll run the `detect_and_correct_compute_addresses_if_unattached` pipeline in the GCP Thrifty mod.

Per the [README](https://github.com/turbot/flowpipe-mod-gcp-thrifty?tab=readme-ov-file#gcp-thrifty-mod-for-flowpipe), the setup entails:

- Cloning the mod

- Visiting its directory and installing dependencies (`flowpipe mod install`)

- Starting Steampipe as a service (`steampipe service start`)

- Using `credential_import` to give Flowpipe access to the credentials in Steampipe's `gcp.spc` 

- Optionally, configuring variables that govern the mod's behavior.

Let's run the pipeline. 

```
$ flowpipe pipeline run detect_and_correct_compute_addresses_if_unattached --arg='approvers=[""]'
```

With no approvers, the pipeline runs without interruptions and reports that both addresses are unattached.

```
Detected unattached Compute address ip-addr-1 [us-central1/encoded-blend-333001].
Detected unattached Compute address ip-addr-2 [us-central1/encoded-blend-333001].
```

Now let's take action on them. 

## Wizard mode for interactive decisions

In wizard mode, Flowpipe prompts you to decide whether to release each address. It's the default, so we can run the pipeline with no arguments.

```
flowpipe pipeline run detect_and_correct_compute_addresses_if_unattached
```

Now, Flowpipe pauses on each detected address, so you can decide whether to release it. We'll choose `Skip` and keep this one for now.

```
┃ Detected unattached Compute address ip1 [us-central1/encoded-blend-333001].
┃ > Skip
┃   Delete Compute Address
```

Note that this interaction can instead happen in Slack, MSTeams, or email by naming — in `approvers` — one or more notifiers configured with alternate [integrations](https://flowpipe.io/docs/reference/config-files/integration).

Flowpipe proceeds to the next detected address. This time we'll choose `Delete Compute Address`; Flowpipe takes that action and reports the outcome.

```
Deleted Compute address ip-addr-2 [us-central1/encoded-blend-333001].
```

Nice! 

## Delete old compute snapshots automatically

You won't want to automatically release IP addresses without checking each one, but you might want to automatically delete compute snapshots older than a threshold number of days. There's another pipeline for that, `detect_and_correct_compute_snapshots_exceeding_max_age`. 

To use that pipeline automatically, you'll activate a [query trigger](https://flowpipe.io/docs/flowpipe-hcl/trigger/query) to query for old snapshots. And you'll set some configuration variables to control how it works. We'll put them into a file, `continuous-compliance.fpvars`.

```hcl
approvers = []
compute_snapshots_exceeding_max_age_trigger_enabled  = true
compute_snapshots_exceeding_max_age_trigger_schedule = "weekly"
compute_snapshots_exceeding_max_age_default_action   = "delete_snapshot"
compute_snapshots_exceeding_max_age_days             = 90
```

Then we'll run Flowpipe in server mode and point it at that package of variables.

```
flowpipe server --var-file=continuous-compliance.fpvars
```

With `approvers = []` the pipeline handles each detected old snapshot without requiring approval. The default action is normally `notify` but here we override to `delete_snapshot`. Now, snapshots older than 90 days will be automatically deleted on a weekly basis.

## Configuration-driven workflow

A detect-and-correct mod like this one doesn't require you to write any SQL or HCL. It's a tool that you configure and then use to perform a range of tasks you'd like to automate, with or without approval. Give GCP Thrifty a try, and please [let us know](https://turbot.com/community/join) how it goes.

## See it in action

<div className="flex justify-center">
<iframe
    class="youtube-video"
    src="https://www.youtube-nocookie.com/embed/v0ODTKP3blQ" 
    frameBorder="0"
    allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture; web-share"
    allowFullScreen
    title="Control GCP costs with 22 workflows for FinOps"
>
</iframe>
</div>

FinOps workflows to discover, approve, and automate GCP cost savings recommendations.

Control GCP costs with 22 workflows for FinOps


Many DevOps challenges follow similar patterns. So we're creating a set of off-the-shelf Flowpipe mods to address these common scenarios head-on, starting with [AWS Thrifty](https://hub.flowpipe.io/mods/turbot/aws_thrifty). For example, suppose you want to detect and release unattached elastic IP addresses. Here are three basic patterns for doing that.

- **Wizard (with approval)**: Run Flowpipe on the command line, detect unattached EIPs, act on the choices `Skip` or `Release`.

- **Scheduled (with approval)**: Run on a schedule, detect, act on the choices `Skip` or `Release`.

- **Scheduled (no approval)**: Run on a schedule, detect and automatically `Release`.

You'll write no code to do any of those things, instead you'll configure mod variables such as:

- The database (Steampipe or other) that feeds the pipeline with the output of [a SQL query](https://hub.powerpipe.io/mods/turbot/aws_thrifty/controls/control.unattached_eips?context=benchmark.network#sql).

- A [notifier](/docs/reference/config-files/notifier) that binds the mod to a communication channel — an [integration](/docs/reference/config-files/integration) — that might be [email](/docs/reference/config-files/integration/email), [Slack](/docs/reference/config-files/integration/slack), or [MS Teams](/docs/reference/config-files/integration/msteams). 

- One more `approvers`, which are notifiers that obtain [action/approval decisions](/blog/human-interaction).

- Whether to run on a [trigger](/blog/query-trigger).

- The default action, e.g. `notify` or `release`.

## Run as a wizard, with Slack-based approval

To run this pipeline using Slack to receive notifications and take actions, [install the mod](https://hub.flowpipe.io/mods/turbot/aws_thrifty), start Steampipe (`steampipe service start`), configure a [Slack integration and notifier](/docs/reference/config-files/integration/slack#setting-up-as-slack-integration), and start the [Flowpipe server](/docs/run/server).

Now run this command.

```bash
flowpipe pipeline run detect_and_correct_vpc_eips_unattached \
  --arg approvers='["slack"]' \
  --arg host=local
```

In this case the pipeline detects one unattached EIP, and waits for a decision to be made in Slack channel.

<div>
  <img width="90%" alt="slack awaiting decision" src="/images/blog/2024-05-devsecops-mod-library/unattached_eip_1.png"/>
</div>

If we choose `Release`, the pipeline calls a [utility pipeline](https://hub.flowpipe.io/mods/turbot/aws/pipelines/aws.pipeline.release_eip) in the [AWS library mod](https://hub.flowpipe.io/mods/turbot/aws) to release the EIP, and the Slack message changes to confirm the action. 

<div>
  <img width="90%" alt="slack confirmed decision" src="/images/blog/2024-05-devsecops-mod-library/unattached_eip_2.png"/>
</div>

If there are more than one unattached EIPs, these Slack interactions will occur serially. To batch them, you can switch the `max_concurrency` from the default, which is 1, to your desired batch size. 

<div>
  <img width="90%" alt="slack confirmed decision" src="/images/blog/2024-05-devsecops-mod-library/unattached_eip_3.png"/>
</div>

Here `max_concurrency` is 3; we've chosen to release the first two but take no action on the third.


## Detect on a schedule, respond with human input

To use this pattern, activate a trigger to run the same query for unattached EIPs, but on a schedule instead of interactively. Because this requires setting a few variables, we'll package them into a file named for this scenario: `scheduled-interactive.fpvars`.

```hcl
notifier="slack"
approvers=["slack"]
vpc_eips_unattached_trigger_enabled=true
vpc_eips_unattached_trigger_schedule="* * * * *"
```

And then run the server, pointing to the `.fpvars` file and the server's public endpoint.

```
flowpipe server --var-file scheduled-interactive.fpvars --base-url=https://feline-just-gorilla.ngrok-free.app/ 
```

Note that if the trigger's query finds an unattached EIP that you choose to `Skip`, it won't be detected again because Flowpipe's [query trigger](/blog/query-trigger) remembers query results it has already seen. But if you change your mind and want to release an EIP that you skipped, you can use the wizard mode to do that. 

## Detect on a schedule, respond automatically

To use this pattern, we'll use a slightly different package of variables: `scheduled-automatic.fpvars`.

```hcl
notifier="slack"
approvers=[]
vpc_eips_unattached_default_action="release" 
vpc_eips_unattached_trigger_enabled=true
vpc_eips_unattached_trigger_schedule="* * * * *"
```

When `approvers` is the empty list, Flowpipe automically use the default action. Normally that's `notify` but here we override to `release`. Now, unattached EIPs detected on each run of the pipeline will be automatically released. It's as if there's a robot clicking the `Release` button for each detected EIP.

## Configuration-driven workflow

A detect-and-corect mod like this one doesn't require you to write any SQL or HCL. It's a tool that you configure and then use to perform a range of tasks you'd like to automate, with or without approval. Give `flowpipe-mod-aws-thrifty` a try, and please [let us know](https://turbot.com/community/join) how it goes.

## See it in action

<div className="flex justify-center">

<iframe
    class="youtube-video"
    src="https://www.youtube-nocookie.com/embed/PH4YUun2lDU" 
    frameBorder="0"
    allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture; web-share"
    allowFullScreen
    title="Control AWS costs with 29 FinOps workflows"
>
</iframe>
</div>



Flowpipe's new AWS Thrifty mod identifies cost-saving possibilities and realizes them with approval or automatically.

Control AWS costs with 29 workflows for FinOps


Workflows can gather — and automatically act on — all kinds of data, but sometimes you need a human in the loop. Here's a query that finds an IAM access key that's active beyond what our 90-day policy allows. 

```sql
select
  access_key_id,
  user_name,
  account_id,
  _ctx ->> 'connection_name' as conn_name
from
  aws_iam_access_key
where
  create_date < now() - interval '90 days'
  and status = 'Active'
```

It's nice to be able to find this non-compliant resource, but you also want to bring it into compliance, and that often requires somebody's input. Now, with the [input step](/docs/flowpipe-hcl/step/input), you can create a pipeline that will pause, ask for human input, and then continue based on that input.

## Require approval to deactivate an expired access key

This [sample workflow](https://hub.flowpipe.io/mods/turbot/deactivate-expired-aws-iam-access-keys-with-approval) kicks off with a [trigger](/blog/query-trigger) that uses that same query. 

```hcl
trigger "query" "list_expired_iam_access_keys" {
  title       = "List Expired IAM Access Keys"
  description = "List IAM access keys older than 90 days."
  database    = var.database
  schedule    = "* * * * *" # Every minute
  primary_key = "access_key_id"

  sql = <<-EOQ
    select
      access_key_id,
      user_name,
      account_id,
      _ctx ->> 'connection_name' as conn_name
    from
      aws_iam_access_key
    where
      create_date < now() - interval '90 days'
      and status = 'Active';
  EOQ

  capture "insert" {
    pipeline = pipeline.deactivate_expired_iam_access_keys_with_approval
    args = {
      access_keys = self.inserted_rows
    }
  }
}
```

The [capture block](/docs/flowpipe-hcl/trigger/query#capture) relays one or more expired keys to a pipeline that uses [for_each](/docs/flowpipe-hcl/step#for_each) to loop over one or more keys, calling a pipeline for each key. Here the input step in the per-key pipeline.

```hcl
step "input" "prompt_deactivate_expired_iam_access_key" {
  notifier = notifier[param.notifier]
  subject  = "Request to deactivate expired IAM access key ${param.access_key.access_key_id} for user ${param.access_key.user_name} [${param.acce
ss_key.account_id}]"
  prompt   = "Do you want to deactivate IAM access key ${param.access_key.access_key_id} belonging to ${param.access_key.user_name} [${param.acce
ss_key.account_id}]?"
  type     = "button"

  option "Deactivate" {
    label = "Deactivate"
    value = "deactivate"
    style = "ok"
  }

  option "Active" {
    label = "Keep expired key active"
    value = "active"
    style = "alert"
  }
}
```

Flowpipe's input step is generic. It defines what the choices are, and how they're represented — in this case, as buttons — but the input step says nothing about how they're rendered. That's just a configuration detail.

## Alternative renderings

Here's how the choices are rendered in email.

<div>
<img alt="buttons in email" src="/images/blog/2024-03-human-interaction/buttons-in-email.png"/>
</div>

And here's how they look in Slack.

<div>
<img alt="buttons in slack" src="/images/blog/2024-03-human-interaction/buttons-in-slack.png"/>
</div>

Buttons aren't the only [input type](/docs/flowpipe-hcl/step/input#input-types), you can also use `text`, `select`, and `multiselect`. Another sample, which [finds S3 buckets](https://hub.flowpipe.io/mods/turbot/add-s3-bucket-cost-center-tags) missing a required tag, then prompts with a hardcoded list of choices, uses `select`.

We'll circle back to the configuration that produces these effects, but first let's walk through the rest of this workflow. 

## Acting on the input

This step, which runs only if the person chose the `Deactivate` option, calls a pipeline [in the AWS mod for Flowpipe](https://hub.flowpipe.io/mods/turbot/aws) to deactivate the key. 

```hcl
step "pipeline" "deactivate_iam_access_key" {
  if       = step.input.prompt_deactivate_expired_iam_access_key.value == "deactivate"
  pipeline = aws.pipeline.update_iam_access_key
  args = {
    cred          = param.access_key.conn_name
    user_name     = param.access_key.user_name
    access_key_id = param.access_key.access_key_id
    status        = "Inactive"
  }
}
```

The next step sends a confirmation message.

```hcl
step "message" "notify_iam_access_key_deactivated" {
  if         = step.input.prompt_deactivate_expired_iam_access_key.value == "deactivate"
  depends_on = [step.pipeline.deactivate_iam_access_key]

  notifier = notifier[param.notifier]
  text     = "Deactivated IAM access key ${param.access_key.access_key_id} for user ${param.access_key.user_name} [${param.access_key.account_id}]"
}
```

The trigger's `var.database` could refer to any supported database — including Steamipe, Postgres, and SQLite — but since we're using the [AWS plugin](https://hub.steampipe.io/plugins/turbot/aws) we'll set that variable to `postgres://steampipe@localhost:9193/steampipe` and start Steampipe as a service.

```
steampipe service start
```

Then we'll start the Flowpipe server, with a base URL matching a webhook we've set up in Slack.

```
flowpipe server --base-url=https://9f6b-157-131-95-205.ngrok-free.app --verbose
```

Here we can see that the query trigger found a stale key and kicked off an interaction with Slack. 

<div>
<img alt="deactivation input started" src="/images/blog/2024-03-human-interaction/deactivation-input-step-started.png"/>
</div>


The pipeline's input step will now wait for a response. When a person presses either button, control returns to the pipeline. If `Deactivate` was the choice, the two steps we've seen will run: perform the deactivation, and send a confirmation message. Otherwise the workflow sends a message confirming that no action was taken. 

## Configure an integration with a notifier for Slack

Along with the `database` variable, this sample mod defines a `notifier` variable which we'll call `expired_access_key`. It's referenced from a config file we've named `integrations.fpc` which includes an [integration](/docs/reference/config-files/integration) that defines how Flowpipe talks to an external system, in this case Slack. It also includes a [notifier](/docs/reference/config-files/notifier) that binds to the integration and provides details specific to it — in this case, the channel (`security`) where the interaction will occur, and to which the confirmation message will be sent. 

```hcl
integration "slack" "default" {
  token = env("SLACK_TOKEN")
}

notifier "expired_access_key" {

  notify {
    integration = integration.slack.default
    channel     = "security"
  }

}
```

None of this configuration is hardwired in the sample mod. To perform the same interaction in email instead of Slack, you'd just write a different set of definitions.

```hcl
integration "email" "default" {
  smtp_tls      = "required"
  smtps_port    = 587
  smtp_host     = "smtp.gmail.com"
  smtp_username = "judell@example.com"
  smtp_password = env("MY_EMAIL_PASSWORD")
  from = "judell@example.com"
}

notifier "expired_access_key" {

  notify {
    integration = integration.email.default
    to = ["security@example.com"]
  }

}
```

You can even include several `notify` blocks in a `notifier` in order to send a message to multiple channels!

## See it in action

<div className="flex justify-center">
<iframe
    class="youtube-video"
    src="https://www.youtube-nocookie.com/embed/zBxNdOxnrRU"
    frameBorder="0"
    allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture; web-share"
    allowFullScreen
    title="Human interaction with Flowpipe"
>
</iframe>
</div>

## Build your own human-in-the-loop workflows

The `integration` and `notifier` blocks work with `input` and `message` steps to define how Flowpipe communicates and interacts with various external systems, so your pipelines can focus on what should happen, not how. It all adds up to a powerful and flexible toolkit for workflows that interact with people. Try building some interactive workflows, and [let us know](https://turbot.com/community/join) how it goes!



Use the new input step to pause a pipeline and bring a person into the loop via Slack or email.

Workflow for DevOps: Approvals and inputs to engage your team


It's handy to run Flowpipe pipelines directly in your console, but pipelines also need to respond to events. <a href="https://flowpipe.io/docs/flowpipe-hcl/trigger">Triggers</a> are the way to do that. An <a href="https://flowpipe.io/docs/flowpipe-hcl/trigger/http">http</a> trigger relays an inbound webhook call so that, for example, a Slack integration can run  a Flowpipe pipeline. A <a href="https://flowpipe.io/docs/flowpipe-hcl/trigger">schedule</a> trigger runs a pipeline periodically. And a <a href="https://flowpipe.io/docs/flowpipe-hcl/trigger/query">query</a> trigger watches a database, firing in response to data changes. 

## A query trigger that watches for new AWS IAM access keys

Here's a workflow that's driven by a query trigger that periodically asks AWS if any new IAM access keys have appeared. If there are any, the trigger calls a notification pipeline.

<div>
<img width="40%" alt="access key workflow graph" src="/images/blog/2024-03-query-trigger/query_trigger_iam.png"/>
</div>

Here's the HCL definition of the trigger.

```hcl
trigger "query" "new_aws_iam_access_key" {
  description       = "Fire when a new key appears, call a notifier"
  database          = "postgres://steampipe@localhost:9193/steampipe"
  schedule          = "* * * * *"
  primary_key       = "access_key_id"
  sql               = <<EOQ
    select access_key_id, account_id, user_name, create_date
    from aws_iam_access_key
    where create_date > now() - interval '2 weeks'
  EOQ

  capture "insert" {
    pipeline = pipeline.notify_new_access_key
    args = {
      rows = self.inserted_rows
    }
  }

}
```

In this case we're using Steampipe, along with the AWS plugin, to query the Amazon API — so the `database` argument is a Postgres connection string. But pipelines can watch other databases including SQLite, MySQL, and DuckDB. And you could even use another distribution of the AWS plugin: a [foreign-data-wrapper extension](https://steampipe.io/blog/2023-12-postgres-extensions) in your own Postgres database, or a [virtual-table extension](https://steampipe.io/blog/2023-12-sqlite-extensions) in SQLite.

We're using a cron expression to run every minute, but could alternatively use a <a href="https://flowpipe.io/docs/flowpipe-hcl/trigger/schedule#arguments">named interval</a> like `"5m"`.

The primary key is optional, if omitted Flowpipe will use a hash of the row to detect change. In this case we're using the `access_key_id`; it's also possible to create a <a href="https://flowpipe.io/docs/flowpipe-hcl/trigger/query#composite-key">composite key</a> by concatenating columns.

The query finds one or more new access keys. The <a href="https://flowpipe.io/docs/flowpipe-hcl/trigger/query#all-capture-types">capture</a> block controls what happens when the query returns new rows. In this case it relays captured rows to a pipeline that sends a notification. You can also have capture blocks that watch for updates and deletes, and you can run a different pipeline for each set of affected rows. 

## Notify on new access keys

Here's the pipeline called by the trigger. The [notifier](/docs/reference/config-files/notifier) doesn't say anything about which communication medium to use, it could as easily be an email or Slack [integration](/docs/reference/config-files/integration), that's purely a configuration detail. 

```hcl
pipeline "notify_new_access_key" {
  description = "Loop over new access keys, send each to notifier."

  param "rows" {
    type = list
  }

  step "message" "notify" {
    notifier = notifier["new_access_key"]
    for_each = param.rows
    text = <<EOQ
      New access key for ${each.value.user_name}, ${each.value.account_id}: 
        ${each.value.access_key_id},  ${each.value.create_date}
      EOQ
  }  
  
}
```

## Run the workflow

As with any scheduled Flowpipe activity, we'll start Flowpipe in server mode

```
flowpipe server
```

Within one minute we can see that Flowpipe detected new data and fired to run the notification pipeline. 

<div>
<img alt="access key console" src="/images/blog/2024-03-query-trigger/access_key_console.png"/>
</div>


Note that Flowpipe watches your code for changes. If you alter the code while the pipeline is running, your changes take effect immediately.

## See it in action

<div className="flex justify-center">
<iframe
    class="youtube-video"
    src="https://www.youtube-nocookie.com/embed/TJZ5NCMohYo"
    frameBorder="0"
    allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture; web-share"
    allowFullScreen
    title="Data-driven DevOps workflows with query triggers and steps"
>
</iframe>
</div>


## Tap into the Flowpipe ecosystem

Flowpipe's query trigger enables powerful workflows that connect to databases and take action when data changes. You can use any of Steampipe's [140 plugins](https://hub.steampipe.io) to query cloud platforms like AWS and Azure, SaaS tools like Jira, security scanners, network tools, and more. Or you can connect to your own data sources. Likewise you can use any of Flowpipe's [library mods](https://hub.flowpipe.io/?type=library) to take actions in Slack, Teams, Email, Jira, AWS, Azure, and elsewhere, or create your own library pipelines that compose with one another and with published mods.

To get started with Flowpipe: [download](/downloads) the tool, follow the [tutorial](/docs), peruse the [library mods](https://hub.flowpipe.io/?type=library), and check out the [samples](https://hub.flowpipe.io/?type=sample). Then create a workflow that alerts on security events, enforces policies, gathers and publishes metrics, or does something else your DevOps team requires, and [let us know](https://turbot.com/community/join) how it goes!




Watch a database, and run a pipeline when a query yields changed results.

Data-driven DevOps workflows with query triggers and steps

## Introducing Flowpipe

Flowpipe is an open-source cloud scripting engine from [Turbot](https://turbot.com) that enables you to:

**Orchestrate your cloud**. Build simple steps into complex workflows. Run and test locally. Compose solutions across clouds using open source mods.

**Connect people and tools**. Connect your cloud data to people and systems using email, chat & APIs. Workflow steps can even run containers, custom functions, and more.

**Respond to events**. Run workflows manually or on a schedule. Trigger pipelines from webhooks or changes in data.

**Use code, not clicks**. Build and deploy DevOps workflows like infrastructure. Code in HCL and deploy from version control.

## Pipelines as code!

Here's a simple example: a two-step pipeline to get your IP address and location:

```hcl
pipeline "learn_flowpipe" {

  # Simple HTTP step to get data from a URL
  step "http" "get_ipv4" {
    url = "https://api.ipify.org?format=json"
  }

  # Run a nested pipeline from the reallyfreegeoip mod
  step "pipeline" "get_geo" {
    pipeline = reallyfreegeoip.pipeline.get_ip_geolocation
    args = {
      # Automatic dependency resolution determines step order
      ip_address = step.http.get_ipv4.response_body.ip
    }
  }

  output "ip_address" {
    value = step.http.get_ipv4.response_body.ip
  }

  output "latitude" {
    value = step.pipeline.get_geo.output.geolocation.latitude
  }

  output "longitude" {
    value = step.pipeline.get_geo.output.geolocation.longitude
  }
}
```

Pipelines are defined and [run](/docs/run/pipelines) on your local machine:

```sh
$ flowpipe pipeline run learn_flowpipe
```

![](/images/blog/2023-12-introducing-flowpipe/geolocation.png)

## Step into cloud scripting

DevOps professionals use a vast repertoire of tools, scripts and processes to get through the day. Flowpipe's step primitives embrace all of them:

| Type | Description |
| --- | --- |
| [container](/docs/flowpipe-hcl/step/container) | Run a Docker container. |
| [email](/docs/flowpipe-hcl/step/email) | Send an email. |
| [function](/docs/flowpipe-hcl/step/function) | Run an AWS Lambda-compatible function. |
| [http](/docs/flowpipe-hcl/step/http) | Make an HTTP request. |
| [pipeline](/docs/flowpipe-hcl/step/pipeline) | Run another Flowpipe pipeline. |
| [query](/docs/flowpipe-hcl/step/query) | Run a SQL query, works great with [Steampipe](https://steampipe.io)! |
| [sleep](/docs/flowpipe-hcl/step/sleep) | Wait for a defined time period. |
| [transform](/docs/flowpipe-hcl/step/transform) | Use HCL functions to transform data . |

For example, here is a [container step](/docs/flowpipe-hcl/step/container) to run the AWS CLI:

```hcl
pipeline "another_pipe" {

  step "container" "aws_s3_ls" {
    image      = "public.ecr.aws/aws-cli/aws-cli"
    cmd        = ["s3", "ls"]
    env = credentials.aws["my_profile"].env
  }

  output "buckets" {
    value = step.container.aws_s3_ls.stdout
  }
}
```

Don't want to run a container? Use a mod or a query step or a function step. There's more than one way to do it, choose the method that best enables you to orchestrate your cloud and coordinate your team.

## Keep control, in good times and bad

Pipelines may start simple, but we know that things get complicated at cloud scale. Flowpipe has you covered with:

- [timeouts, retries and error handling](/docs/build/write-pipelines/errors)
- [conditional logic](/docs/build/write-pipelines/conditionals)
- [iteration](/docs/build/write-pipelines/iteration) and [parallelism](/docs/build/write-pipelines/control-flow#control-flow)

For example, `for_each` will run a step multiple times in parallel for a collection:

```hcl
step "http" "add_a_user" {
  for_each = ["Jerry", "Elaine", "Newman"]
  url      = "https://myapi.local/api/v1/user"
  method   = "post"
  request_body = jsonencode({
    user_name = "${each.value}"
  })
}
```

And `retry` allows you to [retry the step](/docs/build/write-pipelines/errors#retry) when an error occurs:

```hcl
step "http" "my_request" {
  url    = "https://myapi.local/subscribe"
  method = "post"
  body   = jsonencode({
    name = param.subscriber
  })
  retry {
    max_attempts = 5
    strategy     = "exponential"
    min_interval = 100
    max_interval = 10000
  }
}
```

## Composable Mods and the Flowpipe Hub

Flowpipe mods are open-source composable pipelines so you can remix and reuse your code — or build on the great work of the community. The [Flowpipe Hub](https://hub.flowpipe.io) is a searchable directory of mods to discover and use. The source code for mods is [available on GitHub](https://github.com/topics/flowpipe-mod) if you'd like to learn or contribute.

[Flowpipe library mods](https://hub.flowpipe.io/?type=library) make it easy to work with common services including <a href="https://hub.flowpipe.io/mods/turbot/aws">AWS</a>, <a href="https://hub.flowpipe.io/mods/turbot/github">GitHub</a>, <a href="https://hub.flowpipe.io/mods/turbot/jira">Jira</a>, <a href="https://hub.flowpipe.io/mods/turbot/slack">Slack</a>, <a href="https://hub.flowpipe.io/mods/turbot/teams">Teams</a>, <a href="https://hub.flowpipe.io/mods/turbot/zendesk">Zendesk</a> ... and many more!

[Flowpipe sample mods](https://hub.flowpipe.io/?type=sample) are ready-to-run samples that demonstrate patterns and use of various library mods.

[Creating your own mod](/docs/build/#initializing-a-mod) is easy:

```sh
mkdir my_mod
cd my_mod
flowpipe mod init
```

[Install the AWS mod](/docs/build/mod-dependencies) as a dependency:

```sh
flowpipe mod install github.com/turbot/flowpipe-mod-aws
```

[Use the dependency](/docs/build/write-pipelines) in a pipeline step:

```sh
vi my_pipeline.fp
```

```hcl
pipeline "my_pipeline" {
  step "pipeline" "describe_ec2_instances" {
    pipeline = aws.pipeline.describe_ec2_instances
    args = {
      instance_type = "t2.micro"
      region        = "us-east-1"
    }
  }
}
```

and then run your pipeline!

## Schedules, Events and Triggers

DevOps is filled with routine work, which Flowpipe is happy to do on a schedule. Just setup a [schedule trigger](/docs/flowpipe-hcl/trigger/schedule) to run a pipeline at regular intervals:

```hcl
trigger "schedule" "daily_3pm" {
  schedule = "* 15 * * *"
  pipeline = pipeline.daily_task
}
```

Events are critical to the world of cloud scripting — we need to respond immediately to code pushes, infrastructure change, Slack messages, etc. So Flowpipe has a [http trigger](/docs/flowpipe-hcl/trigger/http) to handle incoming webhooks and run a pipeline:

```hcl
trigger "http" "my_webhook" {
  pipeline = pipeline.my_pipeline
  args     = {
    event = self.request_body
  }
}
```

## Cloud scripting is the sweet spot for DevOps automation

Flowpipe is built for DevOps teams. Express pipelines, steps, and triggers in HCL — the familiar DevOps language. Compose pipelines using mods, and mix in SQL, Python functions, or containers as needed. Develop, test and run it all on your local machine — no deployment surprises or long debug cycles. Then schedule the pipelines or respond to events in real-time.

To get started with Flowpipe: [download](/downloads) the tool, follow the [tutorial](/docs), peruse the [library mods](https://hub.flowpipe.io/?type=library), and check out the [samples](https://hub.flowpipe.io/?type=sample). Then [let us know](https://turbot.com/community/join) how it goes!

Automation and workflow to connect your clouds to the people, systems and data that matter.

Flowpipe: The cloud scripting engine


<table>

<tr>
  <td style={{"width":"7em","fontSize":"smaller"}}>2024-09-03</td>
  <td><a href="https://aws.amazon.com/blogs/infrastructure-and-automation/build-workflows-as-code-automations-for-aws-services-using-flowpipe/">Build workflows-as-code automations for AWS services using Flowpipe</a></td>
  <td><i>aws.amazon.com/blogs</i></td>
</tr>

<tr>
  <td style={{"width":"7em","fontSize":"smaller"}}>2024-01-14</td>
  <td><a href="https://jreyesr.github.io/posts/flowpipe-intro">Introduction to Flowpipe</a></td>
  <td><i>jreyesr.github.io</i></td>
</tr>

<tr>
  <td style={{"width":"7em","fontSize":"smaller"}}>2024-01-11</td>
  <td><a href="https://www.infoworld.com/article/3712125/flowpipe-a-workflow-engine-for-devops-scripters.html">Flowpipe: A workflow engine for devops scripters</a></td>
  <td><i>infoworld.com</i></td>
</tr>

<tr>
  <td style={{"width":"7em","fontSize":"smaller"}}>2023-12-21</td>
  <td><a href="https://www.eitr.tech/blog/2023/12/21/exploring-the-fluidity-of-flowpipe">Exploring the Fluidity of FlowPipe</a></td>
  <td><i>eitr.tech</i></td>
</tr>


</table>



Articles about Flowpipe published elsewhere

Flowpipe

Turbot