feat: add workspace specific tags

backend/windmill-api/src/jobs.rs

@@ -11,6 +11,7 @@ use crate::{
     users::{check_scopes, require_owner_of_path, Authed, OptAuthed},
     utils::require_super_admin,
     variables::get_workspace_key,
+    workers::{CUSTOM_TAGS, CUSTOM_TAGS_PER_WORKSPACE},
     BASE_URL,
 };
 use anyhow::Context;
@@ -1403,6 +1404,33 @@ fn add_raw_string(
     return args;
 }
 
+fn check_tag_available_for_workspace(w_id: &str, tag: &Option<String>) -> error::Result<()> {
+    if let Some(tag) = tag {
+        if tag == "" {
+            return Ok(());
+        }
+        let custom_tags_per_w = &*CUSTOM_TAGS_PER_WORKSPACE;
+        if custom_tags_per_w.0.contains(&tag.to_string()) {
+            Ok(())
+        } else if custom_tags_per_w.1.contains_key(tag)
+            && custom_tags_per_w
+                .1
+                .get(tag)
+                .unwrap()
+                .contains(&w_id.to_string())
+        {
+            Ok(())
+        } else {
+            return Err(error::Error::BadRequest(format!(
+                "Tag {tag} cannot be used on workspace {w_id}: (CUSTOM_TAGS: {:?})",
+                *CUSTOM_TAGS
+            )));
+        }
+    } else {
+        Ok(())
+    }
+}
+
 pub async fn run_flow_by_path(
     authed: Authed,
     Extension(user_db): Extension<UserDB>,
@@ -1424,6 +1452,7 @@ pub async fn run_flow_by_path(
     .fetch_optional(&mut tx)
     .await?
     .flatten();
+    check_tag_available_for_workspace(&w_id, &tag)?;
     let scheduled_for = run_query.get_scheduled_for(tx.transaction_mut()).await?;
     let args = run_query.add_include_headers(headers, args.unwrap_or_default());
     let args = add_raw_string(raw_string, args);
@@ -1469,6 +1498,7 @@ pub async fn run_job_by_path(
     let scheduled_for = run_query.get_scheduled_for(tx.transaction_mut()).await?;
     let args = run_query.add_include_headers(headers, args.unwrap_or_default());
     let args = add_raw_string(raw_string, args);
+    check_tag_available_for_workspace(&w_id, &tag)?;
 
     let (uuid, tx) = push(
         tx,
@@ -1669,6 +1699,7 @@ pub async fn run_wait_result_job_by_path_get(
     let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.clone().begin(&authed).await?).into();
     let (job_payload, tag) =
         script_path_to_payload(script_path, tx.transaction_mut(), &w_id).await?;
+    check_tag_available_for_workspace(&w_id, &tag)?;
 
     let (uuid, tx) = push(
         tx,
@@ -1798,6 +1829,7 @@ async fn run_wait_result_script_by_path_internal(
 
     let args = run_query.add_include_headers(headers, args.unwrap_or_default());
     let args = add_raw_string(raw_string, args);
+    check_tag_available_for_workspace(&w_id, &tag)?;
 
     let (uuid, tx) = push(
         tx,
@@ -1851,6 +1883,7 @@ pub async fn run_wait_result_script_by_hash(
 
     let args = run_query.add_include_headers(headers, args.unwrap_or_default());
     let args = add_raw_string(raw_string, args);
+    check_tag_available_for_workspace(&w_id, &tag)?;
 
     let (uuid, tx) = push(
         tx,
@@ -1951,6 +1984,15 @@ async fn run_wait_result_flow_by_path_internal(
     let scheduled_for = run_query.get_scheduled_for(tx.transaction_mut()).await?;
     let args = run_query.add_include_headers(headers, args.unwrap_or_default());
     let args = add_raw_string(raw_string, args);
+    let tag = sqlx::query_scalar!(
+        "SELECT tag from flow WHERE path = $1 and workspace_id = $2",
+        flow_path,
+        w_id
+    )
+    .fetch_optional(&mut tx)
+    .await?
+    .flatten();
+    check_tag_available_for_workspace(&w_id, &tag)?;
 
     let (uuid, tx) = push(
         tx,
@@ -1969,7 +2011,7 @@ async fn run_wait_result_flow_by_path_internal(
         false,
         None,
         !run_query.invisible_to_owner.unwrap_or(false),
-        None,
+        tag,
     )
     .await?;
     tx.commit().await?;
@@ -2002,6 +2044,7 @@ async fn run_preview_job(
     let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.begin(&authed).await?).into();
     let scheduled_for = run_query.get_scheduled_for(tx.transaction_mut()).await?;
     let args = run_query.add_include_headers(headers, preview.args.unwrap_or_default());
+    check_tag_available_for_workspace(&w_id, &preview.tag)?;
 
     let (uuid, tx) = push(
         tx,
@@ -2096,6 +2139,7 @@ async fn run_preview_flow_job(
     let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.begin(&authed).await?).into();
     let scheduled_for = run_query.get_scheduled_for(tx.transaction_mut()).await?;
     let args = run_query.add_include_headers(headers, raw_flow.args.unwrap_or_default());
+    check_tag_available_for_workspace(&w_id, &raw_flow.tag)?;
 
     let (uuid, tx) = push(
         tx,
@@ -2140,6 +2184,7 @@ pub async fn run_job_by_hash(
     let scheduled_for = run_query.get_scheduled_for(tx.transaction_mut()).await?;
     let args = run_query.add_include_headers(headers, args.unwrap_or_default());
     let args = add_raw_string(raw_string, args);
+    check_tag_available_for_workspace(&w_id, &tag)?;
 
     let (uuid, tx) = push(
         tx,

backend/windmill-api/src/lib.rs

@@ -9,6 +9,7 @@
 use crate::oauth2::AllClients;
 use crate::saml::{SamlSsoLogin, ServiceProviderExt};
 use crate::scim::has_scim_token;
+use crate::tracing_init::MyOnFailure;
 use crate::{
     db::UserDB,
     oauth2::{build_oauth_clients, SlackVerifier},
@@ -16,6 +17,7 @@ use crate::{
     users::{Authed, OptAuthed},
     webhook_util::WebhookShared,
 };
+use anyhow::Context;
 use argon2::Argon2;
 use axum::extract::DefaultBodyLimit;
 use axum::{middleware::from_extractor, routing::get, Extension, Router};
@@ -160,7 +162,8 @@ pub async fn run_server(
             TraceLayer::new_for_http()
                 .on_response(MyOnResponse {})
                 .make_span_with(MyMakeSpan {})
-                .on_request(()),
+                .on_request(())
+                .on_failure(MyOnFailure {}),
         )
         .layer(Extension(db.clone()))
         .layer(Extension(rsmq))
@@ -296,7 +299,8 @@ async fn is_up_to_date() -> Result<String, AppError> {
     let version = HTTP_CLIENT
         .get("https://api.github.com/repos/windmill-labs/windmill/releases/latest")
         .send()
-        .await?
+        .await
+        .context("Impossible to reach api.github")?
         .json::<serde_json::Value>()
         .await?
         .get("tag_name")

backend/windmill-api/src/scim.rs

@@ -1,3 +1,5 @@
+#![allow(non_snake_case)]
+
 /*
  * Author: Ruben Fiszel
  * Copyright: Windmill Labs, Inc 2023
@@ -7,11 +9,11 @@
  */
 
 use axum::{
-    extract::{Path, Query},
+    extract::Query,
     middleware::Next,
     response::{IntoResponse, Response},
-    routing::{get, post},
-    Extension, Json, Router,
+    routing::get,
+    Extension, Router,
 };
 use bytes::{BufMut, BytesMut};
 use hyper::{header, http::HeaderValue, Request, StatusCode};
@@ -19,10 +21,16 @@ use mime_guess::mime;
 use serde::{Deserialize, Serialize};
 use serde_json::json;
 use sql_builder::SqlBuilder;
-use windmill_common::{
-    error::{Error, Result},
-    utils::not_found_if_none,
-};
+use windmill_common::error::{Error, Result};
+
+#[cfg(feature = "enterprise")]
+use axum::{extract::Path, Json};
+
+#[cfg(feature = "enterprise")]
+use windmill_common::utils::not_found_if_none;
+
+#[cfg(feature = "enterprise")]
+use axum::routing::post;
 
 use crate::db::DB;
 
@@ -78,8 +86,6 @@ pub async fn has_scim_token<B>(request: Request<B>, next: Next<B>) -> Response {
         .into_response();
 }
 
-pub type JsonScimResult<T> = std::result::Result<JsonScim<T>, Error>;
-
 impl<T> IntoResponse for JsonScim<T>
 where
     T: Serialize,
@@ -170,11 +176,12 @@ pub async fn get_users(
     ))
 }
 
+#[cfg(feature = "enterprise")]
 #[derive(Deserialize, Debug)]
 pub struct CreateUser {
     userName: String,
 }
-// #[cfg(feature = "enterprise")]
+#[cfg(feature = "enterprise")]
 pub async fn create_user(
     Extension(db): Extension<DB>,
     Json(body): Json<CreateUser>,
@@ -382,6 +389,7 @@ pub async fn update_group(
     }
 }
 
+#[cfg(feature = "enterprise")]
 pub async fn delete_group(Extension(db): Extension<DB>, Path(id): Path<String>) -> Result<()> {
     tracing::info!("SCIM delete group: {:?}", id);
     sqlx::query!("DELETE FROM email_to_igroup WHERE igroup = $1", id)
@@ -393,6 +401,7 @@ pub async fn delete_group(Extension(db): Extension<DB>, Path(id): Path<String>)
     Ok(())
 }
 
+#[cfg(feature = "enterprise")]
 fn convert_name(name: &str) -> String {
     name.replace(" ", "_").to_lowercase()
 }

backend/windmill-api/src/scripts.rs

@@ -45,13 +45,6 @@ use windmill_common::{
 };
 use windmill_queue::{self, schedule::push_scheduled_job, QueueTransaction};
 
-lazy_static::lazy_static! {
-    pub static ref CUSTOM_TAGS: Vec<String> = std::env::var("CUSTOM_TAGS")
-        .ok()
-        .map(|x| x.split(',').map(|x| x.to_string()).filter(|x| !x.is_empty()).collect::<Vec<_>>()).unwrap_or_default();
-
-}
-
 const MAX_HASH_HISTORY_LENGTH_STORED: usize = 20;
 
 #[derive(Serialize, sqlx::FromRow)]

backend/windmill-api/src/tracing_init.rs

@@ -8,7 +8,7 @@
 
 use ::tracing::{field, Span};
 use hyper::Response;
-use tower_http::trace::{MakeSpan, OnResponse};
+use tower_http::trace::{MakeSpan, OnFailure, OnResponse};
 
 #[derive(Clone)]
 pub struct MyOnResponse {}
@@ -28,6 +28,14 @@ impl<B> OnResponse<B> for MyOnResponse {
     }
 }
 
+#[derive(Clone)]
+pub struct MyOnFailure {}
+
+impl<B> OnFailure<B> for MyOnFailure {
+    fn on_failure(&mut self, _b: B, _latency: std::time::Duration, _span: &tracing::Span) {
+        // tracing::error!(latency = latency.as_millis(), "response")
+    }
+}
 #[derive(Clone)]
 pub struct MyMakeSpan {}
 

backend/windmill-api/src/workers.rs

@@ -13,6 +13,8 @@ use axum::{
     Json, Router,
 };
 
+use itertools::Itertools;
+use regex::Regex;
 use serde::{Deserialize, Serialize};
 use sqlx::FromRow;
 use windmill_common::{
@@ -20,6 +22,7 @@ use windmill_common::{
     utils::{paginate, Pagination},
 };
 
+use std::collections::HashMap;
 #[cfg(feature = "benchmark")]
 use std::sync::atomic::Ordering;
 #[cfg(feature = "benchmark")]
@@ -45,6 +48,31 @@ lazy_static::lazy_static! {
         .ok()
         .map(|x| x.split(',').map(|x| x.to_string()).collect::<Vec<_>>()).unwrap_or_default();
 
+    pub static ref CUSTOM_TAGS_PER_WORKSPACE: (Vec<String>, HashMap<String, Vec<String>>) =  process_custom_tags(std::env::var("CUSTOM_TAGS")
+        .ok());
+
+    pub static ref ALL_TAGS: Vec<String> = [CUSTOM_TAGS_PER_WORKSPACE.0.clone(), CUSTOM_TAGS_PER_WORKSPACE.1.keys().map(|x| x.to_string()).collect_vec()].concat();
+
+}
+
+fn process_custom_tags(o: Option<String>) -> (Vec<String>, HashMap<String, Vec<String>>) {
+    let regex = Regex::new(r"^(\w+)\(((?:\w+)\+?)+\)$").unwrap();
+    if let Some(s) = o {
+        let mut global = vec![];
+        let mut specific: HashMap<String, Vec<String>> = HashMap::new();
+        for e in s.split(",") {
+            if let Some(cap) = regex.captures(e) {
+                let tag = cap.get(1).unwrap().as_str().to_string();
+                let workspaces = cap.get(2).unwrap().as_str().split("+");
+                specific.insert(tag, workspaces.map(|x| x.to_string()).collect_vec());
+            } else {
+                global.push(e.to_string());
+            }
+        }
+        (global, specific)
+    } else {
+        (vec![], HashMap::new())
+    }
 }
 
 #[derive(FromRow, Serialize, Deserialize)]
@@ -91,5 +119,5 @@ async fn toggle(Query(query): Query<EnableWorkerQuery>) -> JsonResult<bool> {
 }
 
 async fn get_custom_tags() -> Json<Vec<String>> {
-    Json(CUSTOM_TAGS.clone())
+    Json(ALL_TAGS.clone())
 }

backend/windmill-common/src/error.rs

@@ -115,11 +115,13 @@ pub struct AppError(anyhow::Error);
 // Tell axum how to convert `AppError` into a response.
 impl IntoResponse for AppError {
     fn into_response(self) -> Response {
-        (
-            StatusCode::INTERNAL_SERVER_ERROR,
-            format!("Something went wrong: {}", self.0),
-        )
-            .into_response()
+        let body = body::boxed(body::Full::from(self.0.to_string()));
+        tracing::error!(error = self.0.to_string());
+        axum::response::Response::builder()
+            .header("Content-Type", "text/plain")
+            .status(StatusCode::INTERNAL_SERVER_ERROR)
+            .body(body)
+            .unwrap()
     }
 }
 

backend/windmill-queue/src/queue_transaction.rs

@@ -1,4 +1,4 @@
-use std::{fmt::{Debug}};
+use std::fmt::Debug;
 
 use futures_core::{future::BoxFuture, stream::BoxStream};
 use rsmq_async::{RedisBytes, RsmqConnection};