Subliminal：可配置的任务管理系统

Subliminal 是一个通用的任务管理系统，它能够存储、检索和执行跨基于谷歌云服务的强大微服务架构的任务类型请求及其执行记录。该项目主要作为一个学习项目，展示了包括以下一些先进技术：

• GCP 和 Docker 基础知识
• Rust 编程范式和设计结构
• 微服务系统概念
• DevOps 工作流程和部署策略

概述

在核心上，Subliminal 充当一个普通的 CRUD（创建、读取、更新、删除）应用程序，旨在简化对存储在数据库中的 Task 对象的各种操作。此外，它还提供了将任务调度到一个或多个执行服务的能力。这些执行器，无论托管在云端还是本地，都会异步执行任务并更新数据存储以进行反映。

Subliminal 的整体结构建立在 subliminal crate 的基础上，可通过 crates.io 访问。此 crate 存放构建 Subliminal 生态系统中的每个服务所需的关键工具。

• 数据存储服务：管理 Firestore DB 中的数据库记录，允许操作任务请求和执行数据。
• API 网关：促进与数据存储服务的交互，并允许用户将任务排队以执行。
• 消息队列：在 API 网关和执行器之间建立异步双向通信。
• 任务执行器：负责“连接”到运行 subliminal 的项目，并从消息队列中执行任务。

flowchart LR
    A[API] <--> B[Datastore]
    A <--> C[Task Message Queue]
    C --> D[Executor]
    D --> E((Worker))
    D --> F((Worker))

Datastore、API 和 Message Queue 服务的实现设计为通用，因此它们不依赖于用户数据。因此，它们可以作为（并且确实）托管在 Docker Hub 上的预配置 Docker 镜像（这里 和 这里）。

# Login and set project
gcloud auth login
gcloud projects create test-project --name="Test Project"
gcloud config set project test-project

# Enable the necessary APIs
gcloud services enable pubsub.googleapis.com
gcloud services enable run.googleapis.com
gcloud services enable firestore.googleapis.com

# Deploy the datastore service using the latest subliminal-datastore revision
gcloud run deploy test-project-datstore --image docker.io/brokenfulcrum/subliminal-datastore:latest --allow-unauthenticated --max-instances 1 --port 8080 --set-env-vars "RUST_LOG=debug" --set-env-vars "GOOGLE_PROJECT_ID=test-project"

# Deploy the API service using the latest subliminal-api revision using the URL of the datastore service
gcloud run deploy test-project-api --image docker.io/brokenfulcrum/subliminal-api:latest --allow-unauthenticated --max-instances 1 --port 8080 --set-env-vars "USE_TLS=true" --set-env-vars "RUST_LOG=debug" --set-env-vars "GOOGLE_PROJECT_ID=test-project" --set-env-vars "DATASTORE_ADDRESS=https://test-project-datstore.run.app:443"

# Create the firestorm DB instance
gcloud firestore databases create --location=us-west2

# Create the PubSub topic + subscription to handle task updates
gcloud pubsub topics create TaskExecutionUpdates
gcloud pubsub subscriptions create TaskExecutionUpdatesSubscription --topic TaskExecutionUpdates --topic-project test-project --push-endpoint https://test-project-api.run.app/task_execution_update

# Create the PubSub topic + subscription for workers to pull tasks from the queue
gcloud pubsub topics create TestTaskExecutionQueue
gcloud pubsub subscriptions create TestTaskExecutionQueueSubscription --topic TestTaskExecutionQueue --topic-project test-project --enable-exactly-once-delivery

特定实现的 服务

与Datastore和API服务不同，Executor服务是针对用户的。这里定义要执行的Task。因此，用户必须实现自己的Executor服务并将其“附加”到他们的subliminal GCP项目中。

幸运的是，通过subliminal crate中的ExecutionNodeBuilder，这使得整个过程变得非常简单。

// 1. Define a task
#[derive(Debug, Serialize, Deserialize, Clone)]
pub struct TestStruct {
    pub test: String,
}

// 2. Implement the `Task` trait on the task struct
impl Task for TestStruct {
    fn execute(&self) -> TaskResultData {
        thread::sleep(std::time::Duration::from_secs(5));
        TaskResultData {
            result_status: TaskStatus::Passed,
            result_data: Some(json!({"test": "Hello World!"})),
        }
    }
}

#[tokio::main]
async fn main() {
    // 3. Define the execution node, mapping the execution channel to the deserialization type
    let node = ExecutionNodeBuilder::new(3, GOOGLE_PROJECT_ID, UPDATES_TOPIC)
        .await
        .with_consumer::<TestStruct>("TestStructExecutionRequests-sub")
        .await;

    // 4. Start the node
    node.build().await.unwrap();
}

这允许用户创建一个Executor，该Executor关联一个consumer，该consumer监视TestStructExecutionRequests-sub订阅上的消息，将接收到的数据反序列化为TestStruct的实例，并将其设置在内部的Dispatcher中执行。