API Routes for Modules

docker-compose-test.yaml

@@ -19,7 +19,7 @@ services:
  ports:
  - 9000
  healthcheck:
- test: [ "CMD", "curl", "-f", "http://localhost:9000/minio/health/live" ]
+ test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"]
  interval: 15s
  timeout: 20s
  retries: 5
@@ -46,7 +46,7 @@ services:
  networks:
  - parseable-internal
  healthcheck:
- test: [ "CMD", "curl", "-f", "http://localhost:8000/api/v1/liveness" ]
+ test: ["CMD", "curl", "-f", "http://localhost:8000/api/v1/liveness"]
  interval: 15s
  timeout: 20s
  retries: 5
@@ -58,13 +58,42 @@ services:
  delay: 20s
  max_attempts: 3
 
+ panorama:
+ image: ghcr.io/aldrinjenson/panorama:main
+ networks:
+ - parseable-internal
+ depends_on:
+ - parseable
+ environment:
+ - PANORAMA_ADDRESS="http:0.0.0.0:5000"
+ - PANORAMA_ADMIN_USERNAME=pano_admin
+ - PANORAMA_ADMIN_PASSWORD=pano_admin
+ - PARSEABLE_USERNAME=parseableadmin # todo: make as env var and use in other jobs
+ - PARSEABLE_PASSWORD=parseableadmin
+ - PARSEABLE_BASE_URL="http://parseable:8000/api/v1"
+ deploy:
+ restart_policy:
+ condition: on-failure
+ delay: 20s
+ max_attempts: 3
+
  quest:
  image: ghcr.io/parseablehq/quest:main
- command: ["load", "http://parseable:8000", "parseableadmin", "parseableadmin", "20"]
+ command: [
+ "load",
+ "http://parseable:8000",
+ "parseableadmin",
+ "parseableadmin",
+ "20",
+ "http:0.0.0.0:5000", # todo: make these into env var
+ pano_admin,
+ pano_admin,
+ ]
  networks:
  - parseable-internal
  depends_on:
  - parseable
+ - panorama
  deploy:
  restart_policy:
  condition: on-failure

server/src/external_service.rs

@@ -0,0 +1,131 @@
+/*
+ * Parseable Server (C) 2022 - 2023 Parseable, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+use std::{
+ collections::HashMap,
+ sync::{Arc, RwLock},
+};
+
+use crate::storage::StorageMetadata;
+use once_cell::sync::OnceCell;
+
+pub static MODULE_REGISTRY: OnceCell<Arc<RwLock<ModuleRegistry>>> = OnceCell::new();
+
+pub fn init(metadata: &StorageMetadata) {
+ let mut registry = ModuleRegistry::default();
+ registry.load_registry(metadata);
+
+ MODULE_REGISTRY
+ .set(Arc::new(RwLock::new(registry)))
+ .expect("Module Registry is only set once");
+}
+
+pub fn global_module_registry() -> Arc<RwLock<ModuleRegistry>> {
+ MODULE_REGISTRY
+ .get()
+ .expect("Module Registry initialized in main")
+ .clone()
+}
+
+#[derive(Debug, Default)]
+pub struct ModuleRegistry {
+ inner: HashMap<String, Registration>,
+}
+
+impl ModuleRegistry {
+ pub fn load_registry(&mut self, metadata: &StorageMetadata) {
+ for (module_name, module) in &metadata.modules {
+ self.inner.insert(module_name.clone(), module.clone());
+ }
+ }
+ pub fn register(&mut self, module_name: String, module: Registration) {
+ self.inner.insert(module_name, module);
+ }
+
+ pub fn get(&self, id: &str) -> Option<&Registration> {
+ self.inner.get(id)
+ }
+
+ pub fn get_keys(&self) -> Vec<String> {
+ self.inner.keys().cloned().collect()
+ }
+
+ pub fn deregister(&mut self, module_id: &str) {
+ self.inner.remove(module_id);
+ }
+}
+
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq, Hash)]
+pub struct StreamConfig {
+ pub path: String,
+}
+
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq, Hash)]
+#[serde(rename_all = "camelCase")]
+pub struct Registration {
+ pub version: String,
+ pub url: url::Url,
+ pub username: String,
+ pub password: String,
+ pub stream_config: StreamConfig,
+ pub routes: Vec<Route>,
+}
+
+impl Registration {
+ pub fn get_module_path(&self, path: &str, method: &http::Method) -> Option<String> {
+ self.routes
+ .iter()
+ .find(|x| x.server_path == path && method.eq(&x.method))
+ .map(|route| route.module_path.clone())
+ }
+}
+
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq, Hash)]
+pub struct DeRegistration {
+ pub id: String,
+}
+
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq, Hash)]
+#[serde(rename_all = "camelCase")]
+pub struct Route {
+ pub server_path: String,
+ pub module_path: String,
+ pub method: Method,
+}
+
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq, Hash)]
+#[serde(rename_all = "UPPERCASE")]
+#[allow(clippy::upper_case_acronyms)]
+pub enum Method {
+ GET,
+ PUT,
+ POST,
+ DELETE,
+}
+
+impl PartialEq<Method> for http::Method {
+ fn eq(&self, other: &Method) -> bool {
+ matches!(
+ (self, other),
+ (&http::Method::GET, &Method::GET)
+ | (&http::Method::PUT, &Method::PUT)
+ | (&http::Method::POST, &Method::POST)
+ | (&http::Method::DELETE, &Method::DELETE)
+ )
+ }
+}

server/src/handlers/http.rs

@@ -33,11 +33,12 @@ use rustls::{Certificate, PrivateKey, ServerConfig};
 use rustls_pemfile::{certs, pkcs8_private_keys};
 
 use crate::option::CONFIG;
-use crate::rbac::role::Action;
+use crate::{external_service, rbac::role::Action};
 
 use self::middleware::{DisAllowRootUser, RouteExt};
 
 mod about;
+mod external;
 mod health_check;
 mod ingest;
 mod llm;
@@ -286,6 +287,23 @@ pub fn configure_routes(
  oauth_api = oauth_api.app_data(web::Data::from(client))
  }
 
+ let external_services = web::scope("modules")
+ .service(resource("").route(web::get().to(external::list_modules)))
+ .service(
+ resource("{module}")
+ .route(web::put().to(external::register))
+ .route(web::delete().to(external::deregister)),
+ )
+ .service(
+ resource("{module}/config/{logstream}")
+ .route(web::get().to(external::get_config))
+ .route(web::put().to(external::put_config)),
+ )
+ .service(resource("{module}/{tail}*").to(external::router))
+ .app_data(web::Data::from(Arc::clone(
+ &external_service::global_module_registry(),
+ )));
+
  // Deny request if username is same as the env variable P_USERNAME.
  cfg.service(
  // Base path "{url}/api/v1"
@@ -329,6 +347,7 @@ pub fn configure_routes(
  .service(user_api)
  .service(llm_query_api)
  .service(oauth_api)
+ .service(external_services)
  .service(role_api),
  )
  // GET "/" ==> Serve the static frontend directory