maestro 1.4.0 adds custom pipeline metrics

The 1.4.0 release of maestro is out, and with it comes the ability to record custom metrics from your pipelines. Execution time and success/failure were already tracked, but there was no good way to answer questions like “how many rows did this pipeline process?” or “how many API calls did it take?” Now there is.

If you haven’t heard of maestro, it’s a pipeline orchestration package. You can learn more about it here.

Get it from CRAN:

install.packages("maestro")

Recording metrics with update_metric()

A metric is any value that can be represented as a single number: rows processed, bytes read, API calls made, attempts on a task. To record one, call update_metric() from inside a pipeline with the name of the metric and its value. maestro automatically attributes the metric to the pipeline (and run) that is currently executing.

An important design point is that metrics are recorded on the side. They are not part of a pipeline’s return value, so they never get passed downstream in a DAG.

Here’s a two-step DAG that extracts some flight data and loads it, recording a few metrics along the way:

#' ./pipelines/metric_example.R
#' @maestroFrequency 1 hour
extract_flights <- function() {
  
  # Pretends to fetch data from an API
  df <- data.frame(
    flight_id = 1:5,
    airline_code = c("UA", "AC", "AC", "AA", "DE"),
    departure_time = as.POSIXct("2026-10-14 12:00:00") + c(100, 450, 750, 1450, 1750)
  )
  
  update_metric("rows_read", nrow(df))
  update_metric("bytes_read", as.numeric(object.size(df)))
  
  df
}

#' @maestroInputs extract_flights
load_flights <- function(.input) {
  
  write.csv(.input, tempfile(fileext = ".csv"))
  
  update_metric("rows_written", nrow(.input))
}

Retrieving metrics with get_metrics()

After running the schedule, get_metrics() returns a long data.frame of every metric recorded. Each row gives the pipeline it came from (pipe_name), the run it was recorded in (run_id), the metric name, and its numeric value.

library(maestro)

schedule <- build_schedule(quiet = TRUE)

status <- run_schedule(
  schedule,
  orch_frequency = "1 hour",
  quiet = TRUE
)

get_metrics(schedule)

# A tibble: 3 × 4
  pipe_name       run_id metric       value
  <chr>           <chr>  <chr>        <dbl>
1 extract_flights vPnJoC rows_read        5
2 extract_flights vPnJoC bytes_read    1648
3 load_flights    ulLl5m rows_written     5

Notice that each pipeline gets its own run_id, even within a DAG. Because the result is a plain data.frame, it’s easy to summarise further or write out to a database or file, which makes it a natural fit for monitoring pipeline health over time.

A couple of other things to know:

• Metrics are only recorded when a pipeline is executed by run_schedule(). Calling update_metric() while developing or testing a pipeline function interactively is harmless: it emits an informational message and does nothing.
• Execution time and success/failure are already available via get_status() and get_schedule(), so there’s no need to record those yourself.

Metrics are per execution

Metrics belong to a single execution of a pipeline. Each time a pipeline runs it gets a new run_id, and its metrics are stored separately from those of earlier runs. Within one execution, calling update_metric() again with the same name overwrites the earlier value (the last write wins).

Running the schedule a second time shows this: the metrics from both runs are retained and distinguished by run_id.

status <- run_schedule(
  schedule,
  orch_frequency = "1 hour",
  quiet = TRUE
)

get_metrics(schedule)

# A tibble: 6 × 4
  pipe_name       run_id metric       value
  <chr>           <chr>  <chr>        <dbl>
1 extract_flights vPnJoC rows_read        5
2 extract_flights vPnJoC bytes_read    1648
3 extract_flights bilN7s rows_read        5
4 extract_flights bilN7s bytes_read    1648
5 load_flights    ulLl5m rows_written     5
6 load_flights    R9iMP0 rows_written     5

Accumulating with increment = TRUE

Sometimes a metric is built up over the course of a pipeline rather than computed once, like counting API calls inside a loop or the number of attempts on a flaky task. Setting increment = TRUE adds value to the metric’s current value instead of overwriting it. If the metric hasn’t been set yet in the current run, the first call simply sets it. A negative value decrements.

#' ./pipelines/metric_increment.R
#' @maestroFrequency 1 day
fetch_paginated <- function() {
  
  results <- list()
  
  for (page in 1:3) {
    attempts <- 0
    
    repeat {
      attempts <- attempts + 1
      update_metric("attempts", 1, increment = TRUE)
      
      # Pretends to call an API that sometimes fails
      ok <- runif(1) > 0.5
      if (ok || attempts >= 3) break
    }
    
    update_metric("api_calls", 1, increment = TRUE)
    results[[page]] <- data.frame(page = page, value = rnorm(10))
  }
  
  df <- do.call(rbind, results)
  
  # Overwriting (the default) is fine for a one-off value
  update_metric("rows_read", nrow(df))
  
  df
}

In this example both api_calls and attempts accumulate. When it finishes, api_calls is 3 (one per page) while attempts is at least 3, depending on how many retries were needed. Each metric still ends up with a single value per pipeline run.

For the full details, see the metrics vignette.

Other recent changes

If you missed the 1.3.x releases, there are a couple of smaller things worth mentioning:

• The new @maestroCascadeTags tag propagates selected metadata tags (label, flags, and/or loglevel) from a pipeline to all of its downstream pipelines in a DAG. With no value, it cascades all three.
• get_slot_usage() gained min_datetime and max_datetime parameters so the time window can be fixed rather than anchored to the current time.

The full list of changes is in the release notes.

As always, happy orchestrating!