Direktiv: 事件驱动的无服务器工作流引擎

Direktiv简介

Direktiv是一个事件驱动的无服务器工作流引擎,专为云原生环境下的编排、集成和自动化而设计。它由Go语言开发,采用Apache 2.0许可证开源。Direktiv的核心是一个状态机,使用容器作为工作流中的函数,并在状态之间传递JSON结构化数据。

Direktiv Logo

Direktiv提供了以下主要特性:

YAML定义: 使用简单的YAML语法定义工作流和子工作流,支持if/else条件判断、错误处理、重试、验证等功能。
无服务器: 可以在工作流中调用多个无服务器函数,并将响应合并修改为单个可消费的函数调用。
事件驱动: 可以捕获系统内部或来自AWS、Github等外部源的事件,并基于事件执行工作流。
JSON输入和状态: 使用JSON作为工作流的输入,并以JSON响应调用者。JSON数据会在工作流的各个状态之间保存。
API网关: 包含API网关,可将工作流作为服务暴露给第三方消费者,并支持身份验证。
CloudEvents支持: 原生支持CNCF的CloudEvents规范。
GitOps方法: 所有配置、服务和工作流都可以从Git同步,Git成为单一事实来源。
可观察性: 集成了Prometheus(指标)、Fluent Bit(日志)和OpenTelemetry(检测和跟踪)。
定期任务: 可以通过cron作业定期调用工作流执行重复任务。
可扩展: Direktiv可以在多个层面进行扩展,支持Kubernetes和Knative的扩展特性。
易于扩展: 可以使用简单的Docker容器添加自定义函数。

快速开始

Direktiv提供了一个包含所有必需组件的Docker容器(仅限Linux)。初次启动可能需要几分钟来下载所有必需的镜像。

docker run --privileged -p 8080:80 -ti direktiv/direktiv-kube

如果inotify实例的上限太低,pod可能会卡在"pending"状态。如有必要,可以使用以下命令增加该限制: sudo sysctl fs.inotify.max_user_instances=4096

如果你不使用Linux,请参考安装说明进行安装。

工作流编写

编写Direktiv工作流非常简单。本质上,工作流是通过不同的状态进行,在这些状态之间传递JSON负载。这些状态有不同的类型,提供条件分支、错误管理等功能。最重要的类型是action状态,它基本上是对容器的调用。

下面是一个简单的单步工作流示例:

direktiv_api: workflow/v1

functions:
- id: request
  image: gcr.io/direktiv/functions/http-request:1.0
  type: knative-workflow

states:
- id: joke 
  type: action
  action:
    function: request
    input: 
      method: GET
      url: https://api.chucknorris.io/jokes/random?category=jq(.category // "dev")
  transform:
    joke: jq(.return[0].result.value)

这个工作流使用http-request函数发送GET请求到Chuck Norris笑话API。它使用JQ语句jq(.category // "dev")在运行时评估类别。如果状态中定义了'category'值,就会在URL中使用;否则,默认使用'dev'。

Direktiv Workflow

自定义函数

虽然Direktiv可以使用任何标准容器,但如果需要,编写自定义函数也很容易。自定义函数的基本要求是一个监听8080端口并接受JSON负载的容器。以下是一个简单的Rust自定义函数示例:

use actix_web::{App, HttpResponse, HttpServer, Responder, post};
use actix_web::web::Json;
use serde::{Deserialize, Serialize};

#[derive(Deserialize)]
struct Input {
    name: String,
}

#[derive(Serialize)]
struct Output {
    greeting: String,
}

#[post("/")]
async fn index(info: Json<Input>) -> impl Responder {
    HttpResponse::Ok().json(Output { greeting: format!("Welcome to Direktiv, {}!", info.name) })
}

#[actix_rt::main]
async fn main() -> std::io::Result<()> {
    HttpServer::new(|| {
        App::new()
            .service(index)
    })
        .bind("0.0.0.0:8080")?
        .run()
        .await
}