Packages
caffeine_lang
5.4.0
6.3.1
6.3.0
6.2.2
6.2.1
6.2.0
6.1.2
6.1.1
6.1.0
6.0.0
5.6.0
5.5.0
5.4.4
5.4.3
5.4.2
5.4.1
5.4.0
5.3.0
5.2.0
5.1.1
5.1.0
5.0.12
5.0.11
5.0.10
5.0.8
5.0.7
5.0.6
5.0.5
5.0.4
5.0.1
5.0.0
4.10.0
4.9.0
4.8.3
4.8.2
4.8.1
4.8.0
4.7.9
4.7.8
4.7.7
4.7.6
4.7.5
4.6.7
4.6.6
4.6.5
4.6.4
4.6.3
4.6.2
4.6.0
4.5.1
4.5.0
4.4.4
4.4.3
4.4.1
4.4.0
4.3.7
4.3.6
3.0.6
3.0.5
3.0.4
3.0.3
3.0.2
3.0.1
3.0.0
2.0.5
2.0.4
2.0.3
2.0.2
2.0.1
2.0.0
1.0.2
1.0.1
0.1.0
0.0.24
0.0.23
0.0.22
0.0.21
0.0.20
0.0.19
0.0.18
0.0.17
0.0.16
0.0.15
0.0.14
0.0.13
0.0.12
0.0.11
0.0.10
0.0.9
0.0.8
0.0.7
0.0.6
0.0.5
0.0.4
0.0.2
0.0.1
A compiler for generating reliability artifacts from service expectation definitions.
Current section
Files
Jump to
Current section
Files
src/caffeine_lang/codegen/datadog_cql.gleam
/// Datadog-specific CQL → Terraform HCL emission.
///
/// CQL itself (parser/printer/resolver) is vendor-neutral; this module is the
/// Datadog adapter that turns a parsed evaluation expression into the HCL
/// block tree consumed by the `datadog_service_level_objective` resource.
import caffeine_lang/errors
import caffeine_query_language/ast
import caffeine_query_language/generator as cql_generator
import caffeine_query_language/parser
import caffeine_query_language/printer
import caffeine_query_language/resolver
import gleam/dict
import gleam/list
import gleam/result
import terra_madre/hcl
/// Represents a single named query for TimeSlice formulas.
pub type NamedQuery {
NamedQuery(name: String, query: String)
}
/// Represents a resolved SLO query, either GoodOverTotal or TimeSlice.
pub type ResolvedSloQuery {
ResolvedGoodOverTotal(numerator: String, denominator: String)
ResolvedTimeSlice(
comparator: String,
interval_seconds: Int,
threshold: Float,
/// The formula expression (e.g., "build_time + deploy_time")
formula_expression: String,
/// List of named indicators referenced by the formula
indicators: List(NamedQuery),
)
}
/// Represents the SLO type for Datadog terraform generation.
pub type SloType {
MetricSlo
TimeSliceSlo
}
/// Resolved SLO with HCL blocks ready for terraform generation.
pub type ResolvedSloHcl {
ResolvedSloHcl(slo_type: SloType, blocks: List(hcl.Block))
}
/// Parse a value expression, resolve to primitive, substitute words,
/// and return the resolved SLO query type.
@internal
pub fn resolve_slo_query_typed(
value_expr: String,
substitutions: dict.Dict(String, String),
) -> Result(ResolvedSloQuery, String) {
case parser.parse_expr(value_expr) {
Error(err) -> Error("Parse error: " <> err)
Ok(exp) ->
case resolver.resolve_primitives(exp) {
Ok(resolver.GoodOverTotal(numerator_exp, denominator_exp)) -> {
let numerator_str =
cql_generator.substitute_words(numerator_exp, substitutions)
|> printer.exp_to_string
let denominator_str =
cql_generator.substitute_words(denominator_exp, substitutions)
|> printer.exp_to_string
Ok(ResolvedGoodOverTotal(numerator_str, denominator_str))
}
Ok(resolver.TimeSlice(comparator, interval_seconds, threshold, query)) -> {
let comparator_str = case comparator {
ast.LessThan -> "<"
ast.LessThanOrEqualTo -> "<="
ast.GreaterThan -> ">"
ast.GreaterThanOrEqualTo -> ">="
}
case parser.parse_expr(query) {
Ok(query_exp) -> {
let words = cql_generator.extract_words(query_exp)
let named_queries =
words
|> list.filter_map(fn(word) {
case dict.get(substitutions, word) {
Ok(resolved) -> Ok(NamedQuery(word, resolved))
Error(_) -> Error(Nil)
}
})
case named_queries {
[] ->
Ok(
ResolvedTimeSlice(
comparator_str,
interval_seconds,
threshold,
"query1",
[NamedQuery("query1", query)],
),
)
_ -> {
let formula_expr = printer.strip_outer_parens(query)
Ok(ResolvedTimeSlice(
comparator_str,
interval_seconds,
threshold,
formula_expr,
named_queries,
))
}
}
}
Error(_) -> {
let resolved_query =
dict.get(substitutions, query) |> result.unwrap(query)
Ok(
ResolvedTimeSlice(
comparator_str,
interval_seconds,
threshold,
"query1",
[NamedQuery("query1", resolved_query)],
),
)
}
}
}
Error(err) -> Error("Resolution error: " <> errors.to_message(err))
}
}
}
/// Parse a value expression, resolve to primitive, substitute words,
/// and return HCL blocks ready for Datadog terraform generation.
@internal
pub fn resolve_slo_to_hcl(
value_expr: String,
substitutions: dict.Dict(String, String),
) -> Result(ResolvedSloHcl, String) {
case resolve_slo_query_typed(value_expr, substitutions) {
Ok(ResolvedGoodOverTotal(numerator, denominator)) -> {
let query_block =
hcl.simple_block("query", [
#("numerator", hcl.StringLiteral(numerator)),
#("denominator", hcl.StringLiteral(denominator)),
])
Ok(ResolvedSloHcl(MetricSlo, [query_block]))
}
Ok(ResolvedTimeSlice(
comparator,
interval_seconds,
threshold,
formula_expression,
named_queries,
)) -> {
let inner_query_blocks =
named_queries
|> list.map(fn(nq) {
let metric_query_block =
hcl.Block(
type_: "metric_query",
labels: [],
attributes: dict.from_list([
#("data_source", hcl.StringLiteral("metrics")),
#("name", hcl.StringLiteral(nq.name)),
#("query", hcl.StringLiteral(nq.query)),
]),
blocks: [],
)
hcl.Block(type_: "query", labels: [], attributes: dict.new(), blocks: [
metric_query_block,
])
})
let formula_block =
hcl.Block(
type_: "formula",
labels: [],
attributes: dict.from_list([
#("formula_expression", hcl.StringLiteral(formula_expression)),
]),
blocks: [],
)
let outer_query_block =
hcl.Block(type_: "query", labels: [], attributes: dict.new(), blocks: [
formula_block,
..inner_query_blocks
])
let time_slice_block =
hcl.Block(
type_: "time_slice",
labels: [],
attributes: dict.from_list([
#("comparator", hcl.StringLiteral(comparator)),
#("query_interval_seconds", hcl.IntLiteral(interval_seconds)),
#("threshold", hcl.FloatLiteral(threshold)),
]),
blocks: [outer_query_block],
)
let sli_specification_block =
hcl.Block(
type_: "sli_specification",
labels: [],
attributes: dict.new(),
blocks: [time_slice_block],
)
Ok(ResolvedSloHcl(TimeSliceSlo, [sli_specification_block]))
}
Error(err) -> Error(err)
}
}