Pipeline clusters (#662)

crates/arroyo-api/src/lib.rs

@@ -1,11 +1,9 @@
-use anyhow::anyhow;
 use axum::response::IntoResponse;
 use axum::Json;
 use cornucopia_async::DatabaseSource;
 use http::StatusCode;
 use serde::{Deserialize, Serialize};
-use std::error::Error;
-use std::net::SocketAddr;
+use std::net::{SocketAddr, TcpListener};
 use time::OffsetDateTime;
 use tonic::transport::Channel;
 use tracing::{error, info};
@@ -38,6 +36,8 @@ use arroyo_rpc::api_types::{checkpoints::*, connections::*, metrics::*, pipeline
 use arroyo_rpc::config::config;
 use arroyo_rpc::formats::*;
 use arroyo_rpc::grpc::compiler_grpc_client::CompilerGrpcClient;
+use arroyo_server_common::shutdown::ShutdownGuard;
+use arroyo_server_common::wrap_start;
 
 mod cloud;
 mod connection_profiles;
@@ -118,25 +118,22 @@ pub async fn compiler_service() -> Result<CompilerGrpcClient<Channel>, ErrorResp
         })
 }
 
-pub async fn start_server(database: DatabaseSource) -> anyhow::Result<()> {
+pub fn start_server(database: DatabaseSource, guard: ShutdownGuard) -> anyhow::Result<u16> {
     let config = config();
     let addr = SocketAddr::new(config.api.bind_address, config.api.http_port);
+    let listener = TcpListener::bind(addr)?;
+    let local_addr = listener.local_addr()?;
 
     let app = rest::create_rest_app(database, &config.controller_endpoint());
 
-    info!("Starting API server on {:?}", addr);
-    axum::Server::bind(&addr)
-        .serve(app.into_make_service())
-        .await
-        .map_err(|e| {
-            anyhow!(
-                "Failed to start API server on {}: {}",
-                addr,
-                e.source()
-                    .map(|e| e.to_string())
-                    .unwrap_or_else(|| e.to_string())
-            )
-        })
+    info!("Starting API server on {:?}", local_addr);
+    guard.into_spawn_task(wrap_start(
+        "api",
+        local_addr,
+        axum::Server::from_tcp(listener)?.serve(app.into_make_service()),
+    ));
+
+    Ok(local_addr.port())
 }
 
 #[derive(Debug, Serialize, Deserialize)]

crates/arroyo-api/src/pipelines.rs

@@ -13,7 +13,7 @@ use petgraph::visit::NodeRef;
 use std::time::Duration;
 
 use crate::{compiler_service, connection_profiles, jobs, types};
-use arroyo_datastream::preview_sink;
+use arroyo_datastream::default_sink;
 use arroyo_rpc::api_types::pipelines::{
     Job, Pipeline, PipelinePatch, PipelinePost, PipelineRestart, QueryValidationResult, StopType,
     ValidateQueryPost,
@@ -297,13 +297,13 @@ pub(crate) async fn create_pipeline_int<'a>(
                 contact support@arroyo.systems for an increase", auth.org_metadata.max_operators)));
     }
 
-    set_parallelism(&mut compiled.program, 1);
+    set_parallelism(&mut compiled.program, req.parallelism as usize);
 
     if is_preview && !config().sinks_in_preview {
         for node in compiled.program.graph.node_weights_mut() {
             // replace all sink connectors with websink for preview
             if node.operator_name == OperatorName::ConnectorSink {
-                node.operator_config = preview_sink().encode_to_vec();
+                node.operator_config = default_sink().encode_to_vec();
             }
         }
     }

crates/arroyo-connectors/src/lib.rs

@@ -7,6 +7,7 @@ use crate::polling_http::PollingHTTPConnector;
 use crate::preview::PreviewConnector;
 use crate::redis::RedisConnector;
 use crate::single_file::SingleFileConnector;
+use crate::stdout::StdoutConnector;
 use crate::webhook::WebhookConnector;
 use anyhow::{anyhow, bail, Context};
 use arroyo_operator::connector::ErasedConnector;
@@ -48,6 +49,7 @@ pub mod preview;
 pub mod redis;
 pub mod single_file;
 pub mod sse;
+pub mod stdout;
 pub mod webhook;
 pub mod websocket;
 
@@ -69,6 +71,7 @@ pub fn connectors() -> HashMap<&'static str, Box<dyn ErasedConnector>> {
         Box::new(RedisConnector {}),
         Box::new(SingleFileConnector {}),
         Box::new(SSEConnector {}),
+        Box::new(StdoutConnector {}),
         Box::new(WebhookConnector {}),
         Box::new(WebsocketConnector {}),
     ];