feat: GitLab VendorBehavior (#8300)

### Description
Adds a GitLab service `VendorBehavior`, so that logs are aggregated into
groups.

This was initially part of #6188.
However, due to the special treatment required for the timestamps
indicating the start and end of log groups, a decision was made to
remove it from the initial task's scope.


### Testing Instructions

1. build turborepo binary
2. push an example monorepo to GitLab
3. run some turbo tasks in GitLab CI
4. check if logs are grouped into sections

---------

Co-authored-by: Chris Olszewski <[email protected]>
Co-authored-by: Chris Olszewski <[email protected]>

crates/turborepo-ci/Cargo.toml

@@ -10,6 +10,7 @@ license = "MIT"
 workspace = true
 
 [dependencies]
+chrono = { workspace = true }
 tracing = { workspace = true }
 
 [dev-dependencies]

crates/turborepo-ci/src/lib.rs

@@ -6,7 +6,10 @@ mod vendors;
 use std::{env, sync::OnceLock};
 
 use crate::vendors::get_vendors;
-pub use crate::{vendor_behavior::VendorBehavior, vendors::Vendor};
+pub use crate::{
+ vendor_behavior::{GroupPrefixFn, VendorBehavior},
+ vendors::Vendor,
+};
 
 static IS_CI: OnceLock<bool> = OnceLock::new();
 static VENDOR: OnceLock<Option<&'static Vendor>> = OnceLock::new();

crates/turborepo-ci/src/vendor_behavior.rs

@@ -1,15 +1,20 @@
-type GroupPrefixFn = fn(group_name: &str) -> String;
+use std::sync::Arc;
+
+use chrono::{DateTime, Utc};
+
+pub type GroupPrefixFn = Arc<dyn Fn(DateTime<Utc>) -> String + Send + Sync>;
+type GroupPrefixFnFactory = fn(group_name: String) -> GroupPrefixFn;
 
 #[derive(Clone, Debug, PartialEq)]
 pub struct VendorBehavior {
- pub group_prefix: GroupPrefixFn,
- pub group_suffix: GroupPrefixFn,
- pub error_group_prefix: Option<GroupPrefixFn>,
- pub error_group_suffix: Option<GroupPrefixFn>,
+ pub group_prefix: GroupPrefixFnFactory,
+ pub group_suffix: GroupPrefixFnFactory,
+ pub error_group_prefix: Option<GroupPrefixFnFactory>,
+ pub error_group_suffix: Option<GroupPrefixFnFactory>,
 }
 
 impl VendorBehavior {
- pub fn new(prefix: GroupPrefixFn, suffix: GroupPrefixFn) -> Self {
+ pub fn new(prefix: GroupPrefixFnFactory, suffix: GroupPrefixFnFactory) -> Self {
  Self {
  group_prefix: prefix,
  group_suffix: suffix,
@@ -18,7 +23,11 @@ impl VendorBehavior {
  }
  }
 
- pub fn with_error(mut self, prefix: GroupPrefixFn, suffix: GroupPrefixFn) -> Self {
+ pub fn with_error(
+ mut self,
+ prefix: GroupPrefixFnFactory,
+ suffix: GroupPrefixFnFactory,
+ ) -> Self {
  self.error_group_prefix = Some(prefix);
  self.error_group_suffix = Some(suffix);
  self

crates/turborepo-ci/src/vendors.rs

@@ -1,4 +1,8 @@
-use std::{collections::HashMap, fmt::Debug, sync::OnceLock};
+use std::{
+ collections::HashMap,
+ fmt::Debug,
+ sync::{Arc, OnceLock},
+};
 
 use crate::vendor_behavior::VendorBehavior;
 
@@ -78,8 +82,8 @@ pub(crate) fn get_vendors() -> &'static [Vendor] {
  branch_env_var: None,
  username_env_var: None,
  behavior: Some(VendorBehavior::new(
- |group_name| format!("##[group]{group_name}\r\n"),
- |_| String::from("##[endgroup]\r\n"),
+ |group_name| Arc::new(move |_| format!("##[group]{group_name}\r\n")),
+ |_| Arc::new(|_| String::from("##[endgroup]\r\n")),
  )),
  },
  Vendor {
@@ -268,12 +272,14 @@ pub(crate) fn get_vendors() -> &'static [Vendor] {
  username_env_var: Some("GITHUB_ACTOR"),
  behavior: Some(
  VendorBehavior::new(
- |group_name| format!("::group::{group_name}\n"),
- |_| String::from("::endgroup::\n"),
+ |group_name| Arc::new(move |_| format!("::group::{group_name}\n")),
+ |_| Arc::new(move |_| String::from("::endgroup::\n")),
  )
  .with_error(
- |group_name| format!("\x1B[;31m{group_name}\x1B[;0m\n"),
- |_| String::new(),
+ |group_name| {
+ Arc::new(move |_| format!("\x1B[;31m{group_name}\x1B[;0m\n"))
+ },
+ |_| Arc::new(|_| String::new()),
  ),
  ),
  },
@@ -288,7 +294,24 @@ pub(crate) fn get_vendors() -> &'static [Vendor] {
  sha_env_var: None,
  branch_env_var: None,
  username_env_var: None,
- behavior: None,
+ // https://docs.gitlab.com/ee/ci/jobs/#custom-collapsible-sections
+ behavior: Some(VendorBehavior::new(
+ |group_name| {
+ Arc::new(move |start_time| {
+ let timestamp = start_time.timestamp();
+ format!(
+ "\\e[0Ksection_start:{timestamp}:{group_name}\\r\\
+ e[0K{group_name}"
+ )
+ })
+ },
+ |group_name| {
+ Arc::new(move |end_time| {
+ let timestamp = end_time.timestamp();
+ format!("\\e[0Ksection_end:{timestamp}:{group_name}\\r\\e[0K")
+ })
+ },
+ )),
  },
  Vendor {
  name: "GoCD",
@@ -553,8 +576,16 @@ pub(crate) fn get_vendors() -> &'static [Vendor] {
  branch_env_var: None,
  username_env_var: None,
  behavior: Some(VendorBehavior::new(
- |group_name| format!("##teamcity[blockOpened name='{group_name}']"),
- |group_name| format!("##teamcity[blockClosed name='{group_name}']"),
+ |group_name| {
+ Arc::new(move |_| {
+ format!("##teamcity[blockOpened name='{group_name}']")
+ })
+ },
+ |group_name| {
+ Arc::new(move |_| {
+ format!("##teamcity[blockClosed name='{group_name}']")
+ })
+ },
  )),
  },
  Vendor {
@@ -569,8 +600,10 @@ pub(crate) fn get_vendors() -> &'static [Vendor] {
  branch_env_var: None,
  username_env_var: None,
  behavior: Some(VendorBehavior::new(
- |group_name| format!("travis_fold:start:{group_name}\r\n"),
- |group_name| format!("travis_fold:end:{group_name}\r\n"),
+ |group_name| {
+ Arc::new(move |_| format!("travis_fold:start:{group_name}\r\n"))
+ },
+ |group_name| Arc::new(move |_| format!("travis_fold:end:{group_name}\r\n")),
  )),
  },
  Vendor {

crates/turborepo-lib/src/task_graph/visitor.rs

@@ -402,15 +402,19 @@ impl<'a> Visitor<'a> {
  } else {
  format!("{}:{}", task_id.package(), task_id.task())
  };
- let (header, footer) = (
-  (vendor_behavior.group_prefix)(&group_name),
-  (vendor_behavior.group_suffix)(&group_name),
- );
- logger.with_header_footer(Some(header), Some(footer));
+
+ let header_factory = (vendor_behavior.group_prefix)(group_name.to_owned());
+ let footer_factory = (vendor_behavior.group_suffix)(group_name.to_owned());
+
+ logger.with_header_footer(Some(header_factory), Some(footer_factory));
 
  let (error_header, error_footer) = (
- vendor_behavior.error_group_prefix.map(|f| f(&group_name)),
- vendor_behavior.error_group_suffix.map(|f| f(&group_name)),
+ vendor_behavior
+ .error_group_prefix
+ .map(|f| f(group_name.to_owned())),
+ vendor_behavior
+ .error_group_suffix
+ .map(|f| f(group_name.to_owned())),
  );
  logger.with_error_header_footer(error_header, error_footer);
  }

crates/turborepo-ui/Cargo.toml

@@ -16,6 +16,7 @@ workspace = true
 
 [dependencies]
 atty = { workspace = true }
+chrono = { workspace = true }
 console = { workspace = true }
 crossterm = "0.27.0"
 dialoguer = { workspace = true }
@@ -27,5 +28,6 @@ thiserror = { workspace = true }
 tracing = { workspace = true }
 tui-term = { workspace = true }
 turbopath = { workspace = true }
+turborepo-ci = { workspace = true }
 turborepo-vt100 = { workspace = true }
 winapi = "0.3.9"

crates/turborepo-ui/src/output.rs

@@ -4,6 +4,8 @@ use std::{
  sync::{Arc, Mutex, RwLock},
 };
 
+use turborepo_ci::GroupPrefixFn;
+
 /// OutputSink represent a sink for outputs that can be written to from multiple
 /// threads through the use of Loggers.
 pub struct OutputSink<W> {
@@ -28,8 +30,8 @@ pub struct OutputClient<W> {
 
 #[derive(Default)]
 struct Marginals {
- header: Option<String>,
- footer: Option<String>,
+ header: Option<GroupPrefixFn>,
+ footer: Option<GroupPrefixFn>,
 }
 
 pub struct OutputWriter<'a, W> {
@@ -95,11 +97,19 @@ impl<W: Write> OutputSink<W> {
 }
 
 impl<W: Write> OutputClient<W> {
- pub fn with_header_footer(&mut self, header: Option<String>, footer: Option<String>) {
+ pub fn with_header_footer(
+ &mut self,
+ header: Option<GroupPrefixFn>,
+ footer: Option<GroupPrefixFn>,
+ ) {
  self.primary = Marginals { header, footer };
  }
 
- pub fn with_error_header_footer(&mut self, header: Option<String>, footer: Option<String>) {
+ pub fn with_error_header_footer(
+ &mut self,
+ header: Option<GroupPrefixFn>,
+ footer: Option<GroupPrefixFn>,
+ ) {
  self.error = Marginals { header, footer };
  }
 
@@ -151,7 +161,8 @@ impl<W: Write> OutputClient<W> {
  // to ensure that the bytes aren't interspersed.
  let mut writers = writers.lock().expect("lock poisoned");
  if let Some(prefix) = header {
- writers.out.write_all(prefix.as_bytes())?;
+ let start_time = chrono::Utc::now();
+ writers.out.write_all(prefix(start_time).as_bytes())?;
  }
  for SinkBytes {
  buffer,
@@ -165,7 +176,8 @@ impl<W: Write> OutputClient<W> {
  writer.write_all(buffer)?;
  }
  if let Some(suffix) = footer {
- writers.out.write_all(suffix.as_bytes())?;
+ let end_time = chrono::Utc::now();
+ writers.out.write_all(suffix(end_time).as_bytes())?;
  }
  }
 
@@ -381,13 +393,19 @@ mod test {
  fn test_marginals() -> io::Result<()> {
  let sink = OutputSink::new(Vec::new(), Vec::new());
  let mut group1_logger = sink.logger(OutputClientBehavior::Grouped);
- group1_logger
- .with_header_footer(Some("good header\n".into()), Some("good footer\n".into()));
- group1_logger
- .with_error_header_footer(Some("bad header\n".into()), Some("bad footer\n".into()));
+ group1_logger.with_header_footer(
+ Some(Arc::new(|_| "good header\n".into())),
+ Some(Arc::new(|_| "good footer\n".into())),
+ );
+ group1_logger.with_error_header_footer(
+ Some(Arc::new(|_| "bad header\n".into())),
+ Some(Arc::new(|_| "bad footer\n".into())),
+ );
  let mut group2_logger = sink.logger(OutputClientBehavior::Grouped);
- group2_logger
- .with_header_footer(Some("good header\n".into()), Some("good footer\n".into()));
+ group2_logger.with_header_footer(
+ Some(Arc::new(|_| "good header\n".into())),
+ Some(Arc::new(|_| "good footer\n".into())),
+ );
 
  let mut group1_out = group1_logger.stdout();
  let mut group2_out = group2_logger.stdout();