Subliminal: Configurable Task Management System

Subliminal is a versatile task management system that enables the storage, retrieval, and execution of task-type requests and their execution records across a robust microservice architecture built on Google Cloud Services. Developed primarily as a learning endeavour, this project showcases a number of advanced technologies, including:

• GCP and Docker Fundamentals
• Rust Programming Paradigms and Design Structures
• Microservice System Concepts
• DevOps Workflow Processes and Deployment Strategies

Overview

At its core, Subliminal functions as a noraml CRUD (Create, Read, Update, Delete) application, designed to facilitate various operations on Task objects stored within a database. Additionally, it provides the capability to dispatch these tasks to one or more executor services. These executors, whether hosted in the cloud or locally, execute tasks while asynchronously updating the datastore for reflection.

The overall structure of Subliminal is built upon the foundation laid within the subliminal crate, accessible via crates.io. This crate houses key utilities necessary for constructing each service within the broader Subliminal ecosystem:

• Datastore Service: Manages database records in Firestore DB, allowing manipulation of task request and execution data.
• API Gateway: Facilitates interaction with the datastore service and allows users to enqueue tasks for execution.
• Message Queue: Establish asynchronous bidirectional communication between the API gateway and executors.
• Task Executors: Responsible for "attaching" to a project running subliminal and executing tasks sourced from the Message Queue

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

The Datastore, API, and Message Queue service implementations are designed to be generic, so they have no dependencies on user data. As such, they can be (and are) hosted as ready-to-run Docker images hosted on Docker Hub (here and here)

# 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

Implementation-specific Services

Unlike the Datastore and API services, the Executor service is specific to the user. This is where they would define the Task to execute. For that reason, the user must implement their own Executor service and "attach" it to their subliminal GCP project

Luckily, this is made very simple through the ExecutionNodeBuilder within the subliminal crate:

// 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();
}

This allows the user to create an Executor with an associated consumer that monitors messages on the TestStructExecutionRequests-sub subscription, deserializing the received data into an instance of TestStruct and setting it for execution within the internal Dispatcher