diff --git a/Cargo.lock b/Cargo.lock index 6580e1de55c..c0a1061dfd1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5272,6 +5272,7 @@ dependencies = [ "rstest", "schemars", "serde", + "serde_json", "strum", "thiserror", "trybuild", diff --git a/openapi/nexus-internal.json b/openapi/nexus-internal.json index c358b4109b5..48b5adcc49a 100644 --- a/openapi/nexus-internal.json +++ b/openapi/nexus-internal.json @@ -2568,9 +2568,60 @@ "datum", "type" ] + }, + { + "type": "object", + "properties": { + "datum": { + "$ref": "#/components/schemas/MissingDatum" + }, + "type": { + "type": "string", + "enum": [ + "missing" + ] + } + }, + "required": [ + "datum", + "type" + ] } ] }, + "DatumType": { + "description": "The type of an individual datum of a metric.", + "type": "string", + "enum": [ + "bool", + "i8", + "u8", + "i16", + "u16", + "i32", + "u32", + "i64", + "u64", + "f32", + "f64", + "string", + "bytes", + "cumulative_i64", + "cumulative_u64", + "cumulative_f32", + "cumulative_f64", + "histogram_i8", + "histogram_u8", + "histogram_i16", + "histogram_u16", + "histogram_i32", + "histogram_u32", + "histogram_i64", + "histogram_u64", + "histogram_f32", + "histogram_f64" + ] + }, "DiskRuntimeState": { "description": "Runtime state of the Disk, which includes its attach state and some minimal metadata", "type": "object", @@ -4128,9 +4179,77 @@ "content", "type" ] + }, + { + "type": "object", + "properties": { + "content": { + "type": "object", + "properties": { + "datum_type": { + "$ref": "#/components/schemas/DatumType" + } + }, + "required": [ + "datum_type" + ] + }, + "type": { + "type": "string", + "enum": [ + "missing_datum_requires_start_time" + ] + } + }, + "required": [ + "content", + "type" + ] + }, + { + "type": "object", + "properties": { + "content": { + "type": "object", + "properties": { + "datum_type": { + "$ref": "#/components/schemas/DatumType" + } + }, + "required": [ + "datum_type" + ] + }, + "type": { + "type": "string", + "enum": [ + "missing_datum_cannot_have_start_time" + ] + } + }, + "required": [ + "content", + "type" + ] } ] }, + "MissingDatum": { + "type": "object", + "properties": { + "datum_type": { + "$ref": "#/components/schemas/DatumType" + }, + "start_time": { + "nullable": true, + "type": "string", + "format": "date-time" + } + }, + "required": [ + "datum_type" + ] + }, "Name": { "title": "A name unique within the parent collection", "description": "Names must begin with a lower case ASCII letter, be composed exclusively of lowercase ASCII, uppercase ASCII, numbers, and '-', and may not end with a '-'. Names cannot be a UUID though they may contain a UUID.", diff --git a/openapi/nexus.json b/openapi/nexus.json index 08e6cd71498..243a0c912b8 100644 --- a/openapi/nexus.json +++ b/openapi/nexus.json @@ -9728,9 +9728,60 @@ "datum", "type" ] + }, + { + "type": "object", + "properties": { + "datum": { + "$ref": "#/components/schemas/MissingDatum" + }, + "type": { + "type": "string", + "enum": [ + "missing" + ] + } + }, + "required": [ + "datum", + "type" + ] } ] }, + "DatumType": { + "description": "The type of an individual datum of a metric.", + "type": "string", + "enum": [ + "bool", + "i8", + "u8", + "i16", + "u16", + "i32", + "u32", + "i64", + "u64", + "f32", + "f64", + "string", + "bytes", + "cumulative_i64", + "cumulative_u64", + "cumulative_f32", + "cumulative_f64", + "histogram_i8", + "histogram_u8", + "histogram_i16", + "histogram_u16", + "histogram_i32", + "histogram_u32", + "histogram_i64", + "histogram_u64", + "histogram_f32", + "histogram_f64" + ] + }, "DerEncodedKeyPair": { "type": "object", "properties": { @@ -12250,6 +12301,22 @@ "items" ] }, + "MissingDatum": { + "type": "object", + "properties": { + "datum_type": { + "$ref": "#/components/schemas/DatumType" + }, + "start_time": { + "nullable": true, + "type": "string", + "format": "date-time" + } + }, + "required": [ + "datum_type" + ] + }, "Name": { "title": "A name unique within the parent collection", "description": "Names must begin with a lower case ASCII letter, be composed exclusively of lowercase ASCII, uppercase ASCII, numbers, and '-', and may not end with a '-'. Names cannot be a UUID though they may contain a UUID.", diff --git a/oximeter/db/notes.txt b/oximeter/db/notes.txt deleted file mode 100644 index 66c3871d460..00000000000 --- a/oximeter/db/notes.txt +++ /dev/null @@ -1,232 +0,0 @@ -Some notes on querying - -For pagination: - -- Timeseries name is enough for paginated list timeseries endpoint. -It's just normal keyset pagination. - -- For the timeseries data, we'll be using limit/offset pagination. We'll -run the query to get the consistent timeseries keys each time. This is -the `ScanParams` part of the `WhichPage`. The `PageSelector` is the offset. - - -Now, how to run more complex queries? A good example is something like, -aggregating the timeseries across all but one field. For example, let's -look at the Nexus HTTP latency data. The fields are: - -- name (String) -- id (Uuid) -- route (String) -- method (String) -- status_code (I64) - -Imagine we wanted to look at the average latency by route, so averaged -across all methods and status codes. (Let's ingore name/id) - -We need to group the timeseries keys by route, to find the set of keys -consistent with each different route. ClickHouse provides the `groupArray` -function, which is an aggregate function that collects multiple values -into an array. So we can do: - -``` -SELECT - field_value, - groupArray(timeseries_key) -FROM fields_string -WHERE field_name = 'route' -GROUP BY field_value; - - -┌─field_value───────────────────────────────────────────┬─groupArray(timeseries_key)────────────────┐ -│ /metrics/producers │ [1916712826069192294,6228796576473532827] │ -│ /metrics/collectors │ [1500085842574282480] │ -│ /metrics/collect/e6bff1ff-24fb-49dc-a54e-c6a350cd4d6c │ [15389669872422126367] │ -│ /sled_agents/fb0f7546-4d46-40ca-9d56-cbb810684ca7 │ [1166666993114742619] │ -└───────────────────────────────────────────────────────┴───────────────────────────────────────────┘ -``` - -This gives an array of timeseries keys where the route is each of the values -on the left. - -So at a very high level, we can average all the timeseries values where the keys -are in each of these different arrays. - - -This kinda works. It produces an array of arrays, the counts for each of the -histograms, grouped by the field value. - -``` -SELECT - field_value, - groupArray(counts) -FROM -( - SELECT - field_value, - timeseries_key - FROM fields_string - WHERE field_name = 'route' -) AS f0 -INNER JOIN -( - SELECT * - FROM measurements_histogramf64 -) AS meas USING (timeseries_key) -GROUP BY field_value -``` - -We can extend this `groupArray(bins), groupArray(counts)` to get both. - - -Ok, we're getting somewhere. The aggregation "combinators" modify the behavior of -aggregations, in pretty suprising and powerful ways. For example: - -``` -SELECT - field_value, - sumForEach(counts) -FROM -( - SELECT - field_value, - timeseries_key - FROM fields_string - WHERE field_name = 'route' -) AS f0 -INNER JOIN -( - SELECT * - FROM measurements_histogramf64 -) AS meas USING (timeseries_key) -GROUP BY field_value -``` - -This applies the `-ForEach` combinator to the sum aggregation. This applies the -aggregation to corresponding elements of a sequence (table?) of arrays. We can -do this with any of the aggregations, `avg`, `min`, etc. - - -The `-Resample` combinator also looks interesting. It uses its arguments to create -a set of intervals, and applies the aggregation within each of those intervals. -So sort of a group-by interval or window function. - -Another useful method is `toStartOfInterval`. This takes a timestamp and an interval, -say 5 seconds, or 10 minutes, and returns the interval into which that timestamp -falls. Could be very helpful for aligning/binning data to time intervals. But -it does "round", in that the bins don't start at the first timestamp, but at -the rounded-down interval from that timestamp. - -It's possible to build intervals that start exactly at the first timestamp with: - -``` -SELECT - timestamp, - toStartOfInterval(timestamp, toIntervalMinute(1)) + ( - SELECT toSecond(min(timestamp)) - FROM measurements_histogramf64 - ) -FROM measurements_histogramf64 -``` - -Or some other rounding shenanigans. - - -Putting lots of this together: - -``` -SELECT - f0.field_name, - f0.field_value, - f1.field_name, - f1.field_value, - minForEach(bins), - avgForEach(counts) -FROM -( - SELECT - field_name, - field_value, - timeseries_key - FROM fields_string - WHERE field_name = 'route' -) AS f0 -INNER JOIN -( - SELECT - field_name, - field_value, - timeseries_key - FROM fields_i64 - WHERE field_name = 'status_code' -) AS f1 ON f0.timeseries_key = f1.timeseries_key -INNER JOIN -( - SELECT * - FROM measurements_histogramf64 -) AS meas ON f1.timeseries_key = meas.timeseries_key -GROUP BY - f0.field_name, - f0.field_value, - f1.field_name, - f1.field_value -``` - -This selects the field name/value, and the bin and average count for each -histogram, grouping by route and status code. - -These inner select statements look similar to the ones we already -implement in `field.as_query`. But in that case we select *, and here we -probably don't want to do that to avoid errors about things not being -in aggregations or group by's. - -This works (or is syntactically valid) for scalars, if we replace the -combinators with their non-combinator version: e.g, `avgForEach` -> `avg`. - - -Other rando thoughts. - -It'd be nice to have the query builder be able to handle all these, but -I'm not sure how worth it that is. For example, I don't even think we need -the timeseries keys in this query. For the fields where we are specifying -a condition, we have subqueries like: - -``` -SELECT * -FROM fields_{TYPE} -WHERE field_name = NAME -AND field_value OP VALUE; -``` - -For ones where we _don't_ care, we just have the first three lines: - -``` -SELECT * -FROM fields_{TYPE} -WHERE field_name = NAME; -``` - -We can join successive entries on timeseries keys. - -For straight SELECT queries, that's pretty much it, like we have currently. -For AGGREGATION queries, we need to - -- Have a group-by for each (field_name, field_value) pair. This is true -even when we're unselective on the field, because we are still taking that -field, and we still need to group the keys accordingly. -- Select the consistent timeseries keys. This is so we can correlate the -results of the aggregation back to the field names/values which we still -get from the key-select query. -- Apply the aggregation to the measurements. For scalars, this just the -aggregation. For histograms, this is the `-Array` or `-ForEach` combinator -for that aggregation, depending on what we're applying. -- ??? to the timestamps? -- some alignment, grouping, subsampling? It seems -this has to come from the aggregation query, because there's not a useful -default. - -Speaking of defaults, how do these functions behave with missing data? -Or more subtly, what happens if two histograms (say) have the same number -of bins, but the actual bin edges are different? ClickHouse itself doesn't -deal with this AFAICT, which means we'd need to do that in the client. -Ah, but that is unlikely, since we're only aggregating data from the -same timeseries, with the same key. So far anyway. I'm not sure what'll -happen when we start correlating data between timeseries. diff --git a/oximeter/db/schema/replicated/4/up01.sql b/oximeter/db/schema/replicated/4/up01.sql new file mode 100644 index 00000000000..f36745ae2e4 --- /dev/null +++ b/oximeter/db/schema/replicated/4/up01.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_bool_local MODIFY COLUMN datum Nullable(UInt8) \ No newline at end of file diff --git a/oximeter/db/schema/replicated/4/up02.sql b/oximeter/db/schema/replicated/4/up02.sql new file mode 100644 index 00000000000..0f76398652f --- /dev/null +++ b/oximeter/db/schema/replicated/4/up02.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_bool MODIFY COLUMN datum Nullable(UInt8) \ No newline at end of file diff --git a/oximeter/db/schema/replicated/4/up03.sql b/oximeter/db/schema/replicated/4/up03.sql new file mode 100644 index 00000000000..175b23d71bd --- /dev/null +++ b/oximeter/db/schema/replicated/4/up03.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_i8_local MODIFY COLUMN datum Nullable(Int8) \ No newline at end of file diff --git a/oximeter/db/schema/replicated/4/up04.sql b/oximeter/db/schema/replicated/4/up04.sql new file mode 100644 index 00000000000..4c8f22d8e6b --- /dev/null +++ b/oximeter/db/schema/replicated/4/up04.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_i8 MODIFY COLUMN datum Nullable(Int8) \ No newline at end of file diff --git a/oximeter/db/schema/replicated/4/up05.sql b/oximeter/db/schema/replicated/4/up05.sql new file mode 100644 index 00000000000..82490a81cac --- /dev/null +++ b/oximeter/db/schema/replicated/4/up05.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_u8_local MODIFY COLUMN datum Nullable(UInt8) \ No newline at end of file diff --git a/oximeter/db/schema/replicated/4/up06.sql b/oximeter/db/schema/replicated/4/up06.sql new file mode 100644 index 00000000000..c6896821274 --- /dev/null +++ b/oximeter/db/schema/replicated/4/up06.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_u8 MODIFY COLUMN datum Nullable(UInt8) \ No newline at end of file diff --git a/oximeter/db/schema/replicated/4/up07.sql b/oximeter/db/schema/replicated/4/up07.sql new file mode 100644 index 00000000000..43eb40515ba --- /dev/null +++ b/oximeter/db/schema/replicated/4/up07.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_i16_local MODIFY COLUMN datum Nullable(Int16) \ No newline at end of file diff --git a/oximeter/db/schema/replicated/4/up08.sql b/oximeter/db/schema/replicated/4/up08.sql new file mode 100644 index 00000000000..1d983a3c839 --- /dev/null +++ b/oximeter/db/schema/replicated/4/up08.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_i16 MODIFY COLUMN datum Nullable(Int16) \ No newline at end of file diff --git a/oximeter/db/schema/replicated/4/up09.sql b/oximeter/db/schema/replicated/4/up09.sql new file mode 100644 index 00000000000..e52c2adf5ff --- /dev/null +++ b/oximeter/db/schema/replicated/4/up09.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_u16_local MODIFY COLUMN datum Nullable(UInt16) \ No newline at end of file diff --git a/oximeter/db/schema/replicated/4/up10.sql b/oximeter/db/schema/replicated/4/up10.sql new file mode 100644 index 00000000000..d8a69fff1ae --- /dev/null +++ b/oximeter/db/schema/replicated/4/up10.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_u16 MODIFY COLUMN datum Nullable(UInt16) \ No newline at end of file diff --git a/oximeter/db/schema/replicated/4/up11.sql b/oximeter/db/schema/replicated/4/up11.sql new file mode 100644 index 00000000000..b3c2d8de922 --- /dev/null +++ b/oximeter/db/schema/replicated/4/up11.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_i32_local MODIFY COLUMN datum Nullable(Int32) \ No newline at end of file diff --git a/oximeter/db/schema/replicated/4/up12.sql b/oximeter/db/schema/replicated/4/up12.sql new file mode 100644 index 00000000000..65fca2e1b21 --- /dev/null +++ b/oximeter/db/schema/replicated/4/up12.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_i32 MODIFY COLUMN datum Nullable(Int32) \ No newline at end of file diff --git a/oximeter/db/schema/replicated/4/up13.sql b/oximeter/db/schema/replicated/4/up13.sql new file mode 100644 index 00000000000..df7c520e358 --- /dev/null +++ b/oximeter/db/schema/replicated/4/up13.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_u32_local MODIFY COLUMN datum Nullable(UInt32) \ No newline at end of file diff --git a/oximeter/db/schema/replicated/4/up14.sql b/oximeter/db/schema/replicated/4/up14.sql new file mode 100644 index 00000000000..a4cb43fb90e --- /dev/null +++ b/oximeter/db/schema/replicated/4/up14.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_u32 MODIFY COLUMN datum Nullable(UInt32) \ No newline at end of file diff --git a/oximeter/db/schema/replicated/4/up15.sql b/oximeter/db/schema/replicated/4/up15.sql new file mode 100644 index 00000000000..f7583dbdee7 --- /dev/null +++ b/oximeter/db/schema/replicated/4/up15.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_i64_local MODIFY COLUMN datum Nullable(Int64) \ No newline at end of file diff --git a/oximeter/db/schema/replicated/4/up16.sql b/oximeter/db/schema/replicated/4/up16.sql new file mode 100644 index 00000000000..b458243d749 --- /dev/null +++ b/oximeter/db/schema/replicated/4/up16.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_i64 MODIFY COLUMN datum Nullable(Int64) \ No newline at end of file diff --git a/oximeter/db/schema/replicated/4/up17.sql b/oximeter/db/schema/replicated/4/up17.sql new file mode 100644 index 00000000000..9229a977049 --- /dev/null +++ b/oximeter/db/schema/replicated/4/up17.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_u64_local MODIFY COLUMN datum Nullable(UInt64) \ No newline at end of file diff --git a/oximeter/db/schema/replicated/4/up18.sql b/oximeter/db/schema/replicated/4/up18.sql new file mode 100644 index 00000000000..6e2a2a51916 --- /dev/null +++ b/oximeter/db/schema/replicated/4/up18.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_u64 MODIFY COLUMN datum Nullable(UInt64) \ No newline at end of file diff --git a/oximeter/db/schema/replicated/4/up19.sql b/oximeter/db/schema/replicated/4/up19.sql new file mode 100644 index 00000000000..8f16b5d41e1 --- /dev/null +++ b/oximeter/db/schema/replicated/4/up19.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_f32_local MODIFY COLUMN datum Nullable(Float32) \ No newline at end of file diff --git a/oximeter/db/schema/replicated/4/up20.sql b/oximeter/db/schema/replicated/4/up20.sql new file mode 100644 index 00000000000..92635927404 --- /dev/null +++ b/oximeter/db/schema/replicated/4/up20.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_f32 MODIFY COLUMN datum Nullable(Float32) \ No newline at end of file diff --git a/oximeter/db/schema/replicated/4/up21.sql b/oximeter/db/schema/replicated/4/up21.sql new file mode 100644 index 00000000000..72abba6216b --- /dev/null +++ b/oximeter/db/schema/replicated/4/up21.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_f64_local MODIFY COLUMN datum Nullable(Float64) \ No newline at end of file diff --git a/oximeter/db/schema/replicated/4/up22.sql b/oximeter/db/schema/replicated/4/up22.sql new file mode 100644 index 00000000000..0d8522bc031 --- /dev/null +++ b/oximeter/db/schema/replicated/4/up22.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_f64 MODIFY COLUMN datum Nullable(Float64) \ No newline at end of file diff --git a/oximeter/db/schema/replicated/4/up23.sql b/oximeter/db/schema/replicated/4/up23.sql new file mode 100644 index 00000000000..96b94c28958 --- /dev/null +++ b/oximeter/db/schema/replicated/4/up23.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_cumulativei64_local MODIFY COLUMN datum Nullable(Int64) \ No newline at end of file diff --git a/oximeter/db/schema/replicated/4/up24.sql b/oximeter/db/schema/replicated/4/up24.sql new file mode 100644 index 00000000000..55df76c25fb --- /dev/null +++ b/oximeter/db/schema/replicated/4/up24.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_cumulativei64 MODIFY COLUMN datum Nullable(Int64) \ No newline at end of file diff --git a/oximeter/db/schema/replicated/4/up25.sql b/oximeter/db/schema/replicated/4/up25.sql new file mode 100644 index 00000000000..fac73694826 --- /dev/null +++ b/oximeter/db/schema/replicated/4/up25.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_cumulativeu64_local MODIFY COLUMN datum Nullable(UInt64) \ No newline at end of file diff --git a/oximeter/db/schema/replicated/4/up26.sql b/oximeter/db/schema/replicated/4/up26.sql new file mode 100644 index 00000000000..182b2b47045 --- /dev/null +++ b/oximeter/db/schema/replicated/4/up26.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_cumulativeu64 MODIFY COLUMN datum Nullable(UInt64) \ No newline at end of file diff --git a/oximeter/db/schema/replicated/4/up27.sql b/oximeter/db/schema/replicated/4/up27.sql new file mode 100644 index 00000000000..b482d00f81f --- /dev/null +++ b/oximeter/db/schema/replicated/4/up27.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_cumulativef32_local MODIFY COLUMN datum Nullable(Float32) \ No newline at end of file diff --git a/oximeter/db/schema/replicated/4/up28.sql b/oximeter/db/schema/replicated/4/up28.sql new file mode 100644 index 00000000000..cefbe563958 --- /dev/null +++ b/oximeter/db/schema/replicated/4/up28.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_cumulativef32 MODIFY COLUMN datum Nullable(Float32) \ No newline at end of file diff --git a/oximeter/db/schema/replicated/4/up29.sql b/oximeter/db/schema/replicated/4/up29.sql new file mode 100644 index 00000000000..59e21f353d6 --- /dev/null +++ b/oximeter/db/schema/replicated/4/up29.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_cumulativef64_local MODIFY COLUMN datum Nullable(Float64) \ No newline at end of file diff --git a/oximeter/db/schema/replicated/4/up30.sql b/oximeter/db/schema/replicated/4/up30.sql new file mode 100644 index 00000000000..a609e6ad3c8 --- /dev/null +++ b/oximeter/db/schema/replicated/4/up30.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_cumulativef64 MODIFY COLUMN datum Nullable(Float64) \ No newline at end of file diff --git a/oximeter/db/schema/replicated/4/up31.sql b/oximeter/db/schema/replicated/4/up31.sql new file mode 100644 index 00000000000..3726895dd04 --- /dev/null +++ b/oximeter/db/schema/replicated/4/up31.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_string_local MODIFY COLUMN datum Nullable(String); diff --git a/oximeter/db/schema/replicated/4/up32.sql b/oximeter/db/schema/replicated/4/up32.sql new file mode 100644 index 00000000000..5a09705e7e4 --- /dev/null +++ b/oximeter/db/schema/replicated/4/up32.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_string MODIFY COLUMN datum Nullable(String); diff --git a/oximeter/db/schema/replicated/db-init.sql b/oximeter/db/schema/replicated/db-init.sql index 4429f41364b..27df02b709c 100644 --- a/oximeter/db/schema/replicated/db-init.sql +++ b/oximeter/db/schema/replicated/db-init.sql @@ -24,7 +24,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_bool_local ON CLUSTER oximeter_ timeseries_name String, timeseries_key UInt64, timestamp DateTime64(9, 'UTC'), - datum UInt8 + datum Nullable(UInt8) ) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/measurements_bool_local', '{replica}') ORDER BY (timeseries_name, timeseries_key, timestamp) @@ -35,7 +35,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_bool ON CLUSTER oximeter_cluste timeseries_name String, timeseries_key UInt64, timestamp DateTime64(9, 'UTC'), - datum UInt8 + datum Nullable(UInt8) ) ENGINE = Distributed('oximeter_cluster', 'oximeter', 'measurements_bool_local', xxHash64(splitByChar(':', timeseries_name)[1])); @@ -44,7 +44,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_i8_local ON CLUSTER oximeter_cl timeseries_name String, timeseries_key UInt64, timestamp DateTime64(9, 'UTC'), - datum Int8 + datum Nullable(Int8) ) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/measurements_i8_local', '{replica}') ORDER BY (timeseries_name, timeseries_key, timestamp) @@ -55,7 +55,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_i8 ON CLUSTER oximeter_cluster timeseries_name String, timeseries_key UInt64, timestamp DateTime64(9, 'UTC'), - datum Int8 + datum Nullable(Int8) ) ENGINE = Distributed('oximeter_cluster', 'oximeter', 'measurements_i8_local', xxHash64(splitByChar(':', timeseries_name)[1])); @@ -64,7 +64,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_u8_local ON CLUSTER oximeter_cl timeseries_name String, timeseries_key UInt64, timestamp DateTime64(9, 'UTC'), - datum UInt8 + datum Nullable(UInt8) ) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/measurements_u8_local', '{replica}') ORDER BY (timeseries_name, timeseries_key, timestamp) @@ -75,7 +75,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_u8 ON CLUSTER oximeter_cluster timeseries_name String, timeseries_key UInt64, timestamp DateTime64(9, 'UTC'), - datum UInt8 + datum Nullable(UInt8) ) ENGINE = Distributed('oximeter_cluster', 'oximeter', 'measurements_u8_local', xxHash64(splitByChar(':', timeseries_name)[1])); @@ -84,7 +84,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_i16_local ON CLUSTER oximeter_c timeseries_name String, timeseries_key UInt64, timestamp DateTime64(9, 'UTC'), - datum Int16 + datum Nullable(Int16) ) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/measurements_i16_local', '{replica}') ORDER BY (timeseries_name, timeseries_key, timestamp) @@ -95,7 +95,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_i16 ON CLUSTER oximeter_cluster timeseries_name String, timeseries_key UInt64, timestamp DateTime64(9, 'UTC'), - datum Int16 + datum Nullable(Int16) ) ENGINE = Distributed('oximeter_cluster', 'oximeter', 'measurements_i16_local', xxHash64(splitByChar(':', timeseries_name)[1])); @@ -104,7 +104,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_u16_local ON CLUSTER oximeter_c timeseries_name String, timeseries_key UInt64, timestamp DateTime64(9, 'UTC'), - datum UInt16 + datum Nullable(UInt16) ) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/measurements_u16_local', '{replica}') ORDER BY (timeseries_name, timeseries_key, timestamp) @@ -115,7 +115,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_u16 ON CLUSTER oximeter_cluster timeseries_name String, timeseries_key UInt64, timestamp DateTime64(9, 'UTC'), - datum UInt16 + datum Nullable(UInt16) ) ENGINE = Distributed('oximeter_cluster', 'oximeter', 'measurements_u16_local', xxHash64(splitByChar(':', timeseries_name)[1])); @@ -124,7 +124,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_i32_local ON CLUSTER oximeter_c timeseries_name String, timeseries_key UInt64, timestamp DateTime64(9, 'UTC'), - datum Int32 + datum Nullable(Int32) ) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/measurements_i32_local', '{replica}') ORDER BY (timeseries_name, timeseries_key, timestamp) @@ -135,7 +135,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_i32 ON CLUSTER oximeter_cluster timeseries_name String, timeseries_key UInt64, timestamp DateTime64(9, 'UTC'), - datum Int32 + datum Nullable(Int32) ) ENGINE = Distributed('oximeter_cluster', 'oximeter', 'measurements_i32_local', xxHash64(splitByChar(':', timeseries_name)[1])); @@ -144,7 +144,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_u32_local ON CLUSTER oximeter_c timeseries_name String, timeseries_key UInt64, timestamp DateTime64(9, 'UTC'), - datum UInt32 + datum Nullable(UInt32) ) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/measurements_u32_local', '{replica}') ORDER BY (timeseries_name, timeseries_key, timestamp) @@ -155,7 +155,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_u32 ON CLUSTER oximeter_cluster timeseries_name String, timeseries_key UInt64, timestamp DateTime64(9, 'UTC'), - datum UInt32 + datum Nullable(UInt32) ) ENGINE = Distributed('oximeter_cluster', 'oximeter', 'measurements_u32_local', xxHash64(splitByChar(':', timeseries_name)[1])); @@ -164,7 +164,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_i64_local ON CLUSTER oximeter_c timeseries_name String, timeseries_key UInt64, timestamp DateTime64(9, 'UTC'), - datum Int64 + datum Nullable(Int64) ) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/measurements_i64_local', '{replica}') ORDER BY (timeseries_name, timeseries_key, timestamp) @@ -175,7 +175,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_i64 ON CLUSTER oximeter_cluster timeseries_name String, timeseries_key UInt64, timestamp DateTime64(9, 'UTC'), - datum Int64 + datum Nullable(Int64) ) ENGINE = Distributed('oximeter_cluster', 'oximeter', 'measurements_i64_local', xxHash64(splitByChar(':', timeseries_name)[1])); @@ -184,7 +184,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_u64_local ON CLUSTER oximeter_c timeseries_name String, timeseries_key UInt64, timestamp DateTime64(9, 'UTC'), - datum UInt64 + datum Nullable(UInt64) ) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/measurements_u64_local', '{replica}') ORDER BY (timeseries_name, timeseries_key, timestamp) @@ -195,7 +195,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_u64 ON CLUSTER oximeter_cluster timeseries_name String, timeseries_key UInt64, timestamp DateTime64(9, 'UTC'), - datum UInt64 + datum Nullable(UInt64) ) ENGINE = Distributed('oximeter_cluster', 'oximeter', 'measurements_u64_local', xxHash64(splitByChar(':', timeseries_name)[1])); @@ -204,7 +204,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_f32_local ON CLUSTER oximeter_c timeseries_name String, timeseries_key UInt64, timestamp DateTime64(9, 'UTC'), - datum Float32 + datum Nullable(Float32) ) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/measurements_f32_local', '{replica}') ORDER BY (timeseries_name, timeseries_key, timestamp) @@ -215,7 +215,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_f32 ON CLUSTER oximeter_cluster timeseries_name String, timeseries_key UInt64, timestamp DateTime64(9, 'UTC'), - datum Float32 + datum Nullable(Float32) ) ENGINE = Distributed('oximeter_cluster', 'oximeter', 'measurements_f32_local', xxHash64(splitByChar(':', timeseries_name)[1])); @@ -224,7 +224,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_f64_local ON CLUSTER oximeter_c timeseries_name String, timeseries_key UInt64, timestamp DateTime64(9, 'UTC'), - datum Float64 + datum Nullable(Float64) ) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/measurements_f64_local', '{replica}') ORDER BY (timeseries_name, timeseries_key, timestamp) @@ -235,7 +235,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_f64 ON CLUSTER oximeter_cluster timeseries_name String, timeseries_key UInt64, timestamp DateTime64(9, 'UTC'), - datum Float64 + datum Nullable(Float64) ) ENGINE = Distributed('oximeter_cluster', 'oximeter', 'measurements_f64_local', xxHash64(splitByChar(':', timeseries_name)[1])); @@ -244,7 +244,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_string_local ON CLUSTER oximete timeseries_name String, timeseries_key UInt64, timestamp DateTime64(9, 'UTC'), - datum String + datum Nullable(String) ) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/measurements_string_local', '{replica}') ORDER BY (timeseries_name, timeseries_key, timestamp) @@ -255,7 +255,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_string ON CLUSTER oximeter_clus timeseries_name String, timeseries_key UInt64, timestamp DateTime64(9, 'UTC'), - datum String + datum Nullable(String) ) ENGINE = Distributed('oximeter_cluster', 'oximeter', 'measurements_string_local', xxHash64(splitByChar(':', timeseries_name)[1])); @@ -285,7 +285,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_cumulativei64_local ON CLUSTER timeseries_key UInt64, start_time DateTime64(9, 'UTC'), timestamp DateTime64(9, 'UTC'), - datum Int64 + datum Nullable(Int64) ) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/measurements_cumulativei64_local', '{replica}') ORDER BY (timeseries_name, timeseries_key, start_time, timestamp) @@ -297,7 +297,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_cumulativei64 ON CLUSTER oximet timeseries_key UInt64, start_time DateTime64(9, 'UTC'), timestamp DateTime64(9, 'UTC'), - datum Int64 + datum Nullable(Int64) ) ENGINE = Distributed('oximeter_cluster', 'oximeter', 'measurements_cumulativei64_local', xxHash64(splitByChar(':', timeseries_name)[1])); @@ -307,7 +307,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_cumulativeu64_local ON CLUSTER timeseries_key UInt64, start_time DateTime64(9, 'UTC'), timestamp DateTime64(9, 'UTC'), - datum UInt64 + datum Nullable(UInt64) ) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/measurements_cumulativeu64_local', '{replica}') ORDER BY (timeseries_name, timeseries_key, start_time, timestamp) @@ -319,7 +319,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_cumulativeu64 ON CLUSTER oximet timeseries_key UInt64, start_time DateTime64(9, 'UTC'), timestamp DateTime64(9, 'UTC'), - datum UInt64 + datum Nullable(UInt64) ) ENGINE = Distributed('oximeter_cluster', 'oximeter', 'measurements_cumulativeu64_local', xxHash64(splitByChar(':', timeseries_name)[1])); @@ -329,7 +329,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_cumulativef32_local ON CLUSTER timeseries_key UInt64, start_time DateTime64(9, 'UTC'), timestamp DateTime64(9, 'UTC'), - datum Float32 + datum Nullable(Float32) ) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/measurements_cumulativef32_local', '{replica}') ORDER BY (timeseries_name, timeseries_key, start_time, timestamp) @@ -341,7 +341,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_cumulativef32 ON CLUSTER oximet timeseries_key UInt64, start_time DateTime64(9, 'UTC'), timestamp DateTime64(9, 'UTC'), - datum Float32 + datum Nullable(Float32) ) ENGINE = Distributed('oximeter_cluster', 'oximeter', 'measurements_cumulativef32_local', xxHash64(splitByChar(':', timeseries_name)[1])); @@ -351,7 +351,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_cumulativef64_local ON CLUSTER timeseries_key UInt64, start_time DateTime64(9, 'UTC'), timestamp DateTime64(9, 'UTC'), - datum Float64 + datum Nullable(Float64) ) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/measurements_cumulativef64_local', '{replica}') ORDER BY (timeseries_name, timeseries_key, start_time, timestamp) @@ -363,7 +363,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_cumulativef64 ON CLUSTER oximet timeseries_key UInt64, start_time DateTime64(9, 'UTC'), timestamp DateTime64(9, 'UTC'), - datum Float64 + datum Nullable(Float64) ) ENGINE = Distributed('oximeter_cluster', 'oximeter', 'measurements_cumulativef64_local', xxHash64(splitByChar(':', timeseries_name)[1])); diff --git a/oximeter/db/schema/single-node/4/up01.sql b/oximeter/db/schema/single-node/4/up01.sql new file mode 100644 index 00000000000..ccccc9c5fb0 --- /dev/null +++ b/oximeter/db/schema/single-node/4/up01.sql @@ -0,0 +1,9 @@ +/* + * To support missing measurements, we are making all scalar datum columns + * Nullable, so that a NULL value (None in Rust) represents a missing datum at + * the provided timestamp. + * + * Note that arrays cannot be made Nullable, so we need to use an empty array as + * the sentinel value implying a missing measurement. + */ +ALTER TABLE oximeter.measurements_bool MODIFY COLUMN datum Nullable(UInt8) diff --git a/oximeter/db/schema/single-node/4/up02.sql b/oximeter/db/schema/single-node/4/up02.sql new file mode 100644 index 00000000000..4c8f22d8e6b --- /dev/null +++ b/oximeter/db/schema/single-node/4/up02.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_i8 MODIFY COLUMN datum Nullable(Int8) \ No newline at end of file diff --git a/oximeter/db/schema/single-node/4/up03.sql b/oximeter/db/schema/single-node/4/up03.sql new file mode 100644 index 00000000000..c6896821274 --- /dev/null +++ b/oximeter/db/schema/single-node/4/up03.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_u8 MODIFY COLUMN datum Nullable(UInt8) \ No newline at end of file diff --git a/oximeter/db/schema/single-node/4/up04.sql b/oximeter/db/schema/single-node/4/up04.sql new file mode 100644 index 00000000000..1d983a3c839 --- /dev/null +++ b/oximeter/db/schema/single-node/4/up04.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_i16 MODIFY COLUMN datum Nullable(Int16) \ No newline at end of file diff --git a/oximeter/db/schema/single-node/4/up05.sql b/oximeter/db/schema/single-node/4/up05.sql new file mode 100644 index 00000000000..d8a69fff1ae --- /dev/null +++ b/oximeter/db/schema/single-node/4/up05.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_u16 MODIFY COLUMN datum Nullable(UInt16) \ No newline at end of file diff --git a/oximeter/db/schema/single-node/4/up06.sql b/oximeter/db/schema/single-node/4/up06.sql new file mode 100644 index 00000000000..65fca2e1b21 --- /dev/null +++ b/oximeter/db/schema/single-node/4/up06.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_i32 MODIFY COLUMN datum Nullable(Int32) \ No newline at end of file diff --git a/oximeter/db/schema/single-node/4/up07.sql b/oximeter/db/schema/single-node/4/up07.sql new file mode 100644 index 00000000000..a4cb43fb90e --- /dev/null +++ b/oximeter/db/schema/single-node/4/up07.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_u32 MODIFY COLUMN datum Nullable(UInt32) \ No newline at end of file diff --git a/oximeter/db/schema/single-node/4/up08.sql b/oximeter/db/schema/single-node/4/up08.sql new file mode 100644 index 00000000000..b458243d749 --- /dev/null +++ b/oximeter/db/schema/single-node/4/up08.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_i64 MODIFY COLUMN datum Nullable(Int64) \ No newline at end of file diff --git a/oximeter/db/schema/single-node/4/up09.sql b/oximeter/db/schema/single-node/4/up09.sql new file mode 100644 index 00000000000..6e2a2a51916 --- /dev/null +++ b/oximeter/db/schema/single-node/4/up09.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_u64 MODIFY COLUMN datum Nullable(UInt64) \ No newline at end of file diff --git a/oximeter/db/schema/single-node/4/up10.sql b/oximeter/db/schema/single-node/4/up10.sql new file mode 100644 index 00000000000..92635927404 --- /dev/null +++ b/oximeter/db/schema/single-node/4/up10.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_f32 MODIFY COLUMN datum Nullable(Float32) \ No newline at end of file diff --git a/oximeter/db/schema/single-node/4/up11.sql b/oximeter/db/schema/single-node/4/up11.sql new file mode 100644 index 00000000000..0d8522bc031 --- /dev/null +++ b/oximeter/db/schema/single-node/4/up11.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_f64 MODIFY COLUMN datum Nullable(Float64) \ No newline at end of file diff --git a/oximeter/db/schema/single-node/4/up12.sql b/oximeter/db/schema/single-node/4/up12.sql new file mode 100644 index 00000000000..55df76c25fb --- /dev/null +++ b/oximeter/db/schema/single-node/4/up12.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_cumulativei64 MODIFY COLUMN datum Nullable(Int64) \ No newline at end of file diff --git a/oximeter/db/schema/single-node/4/up13.sql b/oximeter/db/schema/single-node/4/up13.sql new file mode 100644 index 00000000000..182b2b47045 --- /dev/null +++ b/oximeter/db/schema/single-node/4/up13.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_cumulativeu64 MODIFY COLUMN datum Nullable(UInt64) \ No newline at end of file diff --git a/oximeter/db/schema/single-node/4/up14.sql b/oximeter/db/schema/single-node/4/up14.sql new file mode 100644 index 00000000000..cefbe563958 --- /dev/null +++ b/oximeter/db/schema/single-node/4/up14.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_cumulativef32 MODIFY COLUMN datum Nullable(Float32) \ No newline at end of file diff --git a/oximeter/db/schema/single-node/4/up15.sql b/oximeter/db/schema/single-node/4/up15.sql new file mode 100644 index 00000000000..a609e6ad3c8 --- /dev/null +++ b/oximeter/db/schema/single-node/4/up15.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_cumulativef64 MODIFY COLUMN datum Nullable(Float64) \ No newline at end of file diff --git a/oximeter/db/schema/single-node/4/up16.sql b/oximeter/db/schema/single-node/4/up16.sql new file mode 100644 index 00000000000..5a09705e7e4 --- /dev/null +++ b/oximeter/db/schema/single-node/4/up16.sql @@ -0,0 +1 @@ +ALTER TABLE oximeter.measurements_string MODIFY COLUMN datum Nullable(String); diff --git a/oximeter/db/schema/single-node/db-init.sql b/oximeter/db/schema/single-node/db-init.sql index ee5e91c4b7e..510c1071c82 100644 --- a/oximeter/db/schema/single-node/db-init.sql +++ b/oximeter/db/schema/single-node/db-init.sql @@ -24,7 +24,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_bool timeseries_name String, timeseries_key UInt64, timestamp DateTime64(9, 'UTC'), - datum UInt8 + datum Nullable(UInt8) ) ENGINE = MergeTree() ORDER BY (timeseries_name, timeseries_key, timestamp) @@ -35,7 +35,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_i8 timeseries_name String, timeseries_key UInt64, timestamp DateTime64(9, 'UTC'), - datum Int8 + datum Nullable(Int8) ) ENGINE = MergeTree() ORDER BY (timeseries_name, timeseries_key, timestamp) @@ -46,7 +46,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_u8 timeseries_name String, timeseries_key UInt64, timestamp DateTime64(9, 'UTC'), - datum UInt8 + datum Nullable(UInt8) ) ENGINE = MergeTree() ORDER BY (timeseries_name, timeseries_key, timestamp) @@ -57,7 +57,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_i16 timeseries_name String, timeseries_key UInt64, timestamp DateTime64(9, 'UTC'), - datum Int16 + datum Nullable(Int16) ) ENGINE = MergeTree() ORDER BY (timeseries_name, timeseries_key, timestamp) @@ -68,7 +68,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_u16 timeseries_name String, timeseries_key UInt64, timestamp DateTime64(9, 'UTC'), - datum UInt16 + datum Nullable(UInt16) ) ENGINE = MergeTree() ORDER BY (timeseries_name, timeseries_key, timestamp) @@ -79,7 +79,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_i32 timeseries_name String, timeseries_key UInt64, timestamp DateTime64(9, 'UTC'), - datum Int32 + datum Nullable(Int32) ) ENGINE = MergeTree() ORDER BY (timeseries_name, timeseries_key, timestamp) @@ -90,7 +90,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_u32 timeseries_name String, timeseries_key UInt64, timestamp DateTime64(9, 'UTC'), - datum UInt32 + datum Nullable(UInt32) ) ENGINE = MergeTree() ORDER BY (timeseries_name, timeseries_key, timestamp) @@ -101,7 +101,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_i64 timeseries_name String, timeseries_key UInt64, timestamp DateTime64(9, 'UTC'), - datum Int64 + datum Nullable(Int64) ) ENGINE = MergeTree() ORDER BY (timeseries_name, timeseries_key, timestamp) @@ -112,7 +112,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_u64 timeseries_name String, timeseries_key UInt64, timestamp DateTime64(9, 'UTC'), - datum UInt64 + datum Nullable(UInt64) ) ENGINE = MergeTree() ORDER BY (timeseries_name, timeseries_key, timestamp) @@ -123,7 +123,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_f32 timeseries_name String, timeseries_key UInt64, timestamp DateTime64(9, 'UTC'), - datum Float32 + datum Nullable(Float32) ) ENGINE = MergeTree() ORDER BY (timeseries_name, timeseries_key, timestamp) @@ -134,7 +134,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_f64 timeseries_name String, timeseries_key UInt64, timestamp DateTime64(9, 'UTC'), - datum Float64 + datum Nullable(Float64) ) ENGINE = MergeTree() ORDER BY (timeseries_name, timeseries_key, timestamp) @@ -145,7 +145,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_string timeseries_name String, timeseries_key UInt64, timestamp DateTime64(9, 'UTC'), - datum String + datum Nullable(String) ) ENGINE = MergeTree() ORDER BY (timeseries_name, timeseries_key, timestamp) @@ -156,6 +156,13 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_bytes timeseries_name String, timeseries_key UInt64, timestamp DateTime64(9, 'UTC'), + /* + * NOTE: Right now we can't unambiguously record a nullable byte array. + * Arrays cannot be nested in `Nullable()` types, and encoding the array as + * a string isn't palatable for a few reasons. + * See: https://github.com/oxidecomputer/omicron/issues/4551 for more + * details. + */ datum Array(UInt8) ) ENGINE = MergeTree() @@ -168,7 +175,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_cumulativei64 timeseries_key UInt64, start_time DateTime64(9, 'UTC'), timestamp DateTime64(9, 'UTC'), - datum Int64 + datum Nullable(Int64) ) ENGINE = MergeTree() ORDER BY (timeseries_name, timeseries_key, start_time, timestamp) @@ -180,7 +187,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_cumulativeu64 timeseries_key UInt64, start_time DateTime64(9, 'UTC'), timestamp DateTime64(9, 'UTC'), - datum UInt64 + datum Nullable(UInt64) ) ENGINE = MergeTree() ORDER BY (timeseries_name, timeseries_key, start_time, timestamp) @@ -192,7 +199,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_cumulativef32 timeseries_key UInt64, start_time DateTime64(9, 'UTC'), timestamp DateTime64(9, 'UTC'), - datum Float32 + datum Nullable(Float32) ) ENGINE = MergeTree() ORDER BY (timeseries_name, timeseries_key, start_time, timestamp) @@ -205,7 +212,7 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_cumulativef64 timeseries_key UInt64, start_time DateTime64(9, 'UTC'), timestamp DateTime64(9, 'UTC'), - datum Float64 + datum Nullable(Float64) ) ENGINE = MergeTree() ORDER BY (timeseries_name, timeseries_key, start_time, timestamp) @@ -217,6 +224,16 @@ CREATE TABLE IF NOT EXISTS oximeter.measurements_histogrami8 timeseries_key UInt64, start_time DateTime64(9, 'UTC'), timestamp DateTime64(9, 'UTC'), + /* + * NOTE: Array types cannot be Nullable, see + * https://clickhouse.com/docs/en/sql-reference/data-types/nullable + * for more details. + * + * This means we need to use empty arrays to indicate a missing value. This + * is unfortunate, and at this point relies on the fact that an + * `oximeter::Histogram` cannot have zero bins. If that changes, we'll need + * to figure out another way to represent missing samples here. + */ bins Array(Int8), counts Array(UInt64) ) diff --git a/oximeter/db/src/client.rs b/oximeter/db/src/client.rs index e1ed06554ca..c8a7db20cbe 100644 --- a/oximeter/db/src/client.rs +++ b/oximeter/db/src/client.rs @@ -1190,7 +1190,7 @@ mod tests { use super::*; use crate::query; use crate::query::field_table_name; - use crate::query::measurement_table_name; + use bytes::Bytes; use chrono::Utc; use omicron_test_utils::dev::clickhouse::{ ClickHouseCluster, ClickHouseInstance, @@ -1198,8 +1198,10 @@ mod tests { use omicron_test_utils::dev::test_setup_log; use oximeter::histogram::Histogram; use oximeter::test_util; + use oximeter::types::MissingDatum; use oximeter::Datum; use oximeter::FieldValue; + use oximeter::Measurement; use oximeter::Metric; use oximeter::Target; use std::net::Ipv6Addr; @@ -2957,76 +2959,102 @@ mod tests { Ok(()) } + async fn test_recall_missing_scalar_measurement_impl( + measurement: Measurement, + client: &Client, + ) -> Result<(), Error> { + let start_time = if measurement.datum().is_cumulative() { + Some(Utc::now()) + } else { + None + }; + let missing_datum = Datum::from( + MissingDatum::new(measurement.datum_type(), start_time).unwrap(), + ); + let missing_measurement = Measurement::new(Utc::now(), missing_datum); + test_recall_measurement_impl(missing_measurement, client).await?; + Ok(()) + } + async fn recall_measurement_bool_test( client: &Client, ) -> Result<(), Error> { let datum = Datum::Bool(true); - let as_json = serde_json::Value::from(1_u64); - test_recall_measurement_impl::(datum, None, as_json, client) + let measurement = Measurement::new(Utc::now(), datum); + test_recall_measurement_impl(measurement.clone(), client).await?; + test_recall_missing_scalar_measurement_impl(measurement, client) .await?; Ok(()) } async fn recall_measurement_i8_test(client: &Client) -> Result<(), Error> { let datum = Datum::I8(1); - let as_json = serde_json::Value::from(1_i8); - test_recall_measurement_impl::(datum, None, as_json, client) + let measurement = Measurement::new(Utc::now(), datum); + test_recall_measurement_impl(measurement.clone(), client).await?; + test_recall_missing_scalar_measurement_impl(measurement, client) .await?; Ok(()) } async fn recall_measurement_u8_test(client: &Client) -> Result<(), Error> { let datum = Datum::U8(1); - let as_json = serde_json::Value::from(1_u8); - test_recall_measurement_impl::(datum, None, as_json, client) + let measurement = Measurement::new(Utc::now(), datum); + test_recall_measurement_impl(measurement.clone(), client).await?; + test_recall_missing_scalar_measurement_impl(measurement, client) .await?; Ok(()) } async fn recall_measurement_i16_test(client: &Client) -> Result<(), Error> { let datum = Datum::I16(1); - let as_json = serde_json::Value::from(1_i16); - test_recall_measurement_impl::(datum, None, as_json, client) + let measurement = Measurement::new(Utc::now(), datum); + test_recall_measurement_impl(measurement.clone(), client).await?; + test_recall_missing_scalar_measurement_impl(measurement, client) .await?; Ok(()) } async fn recall_measurement_u16_test(client: &Client) -> Result<(), Error> { let datum = Datum::U16(1); - let as_json = serde_json::Value::from(1_u16); - test_recall_measurement_impl::(datum, None, as_json, client) + let measurement = Measurement::new(Utc::now(), datum); + test_recall_measurement_impl(measurement.clone(), client).await?; + test_recall_missing_scalar_measurement_impl(measurement, client) .await?; Ok(()) } async fn recall_measurement_i32_test(client: &Client) -> Result<(), Error> { let datum = Datum::I32(1); - let as_json = serde_json::Value::from(1_i32); - test_recall_measurement_impl::(datum, None, as_json, client) + let measurement = Measurement::new(Utc::now(), datum); + test_recall_measurement_impl(measurement.clone(), client).await?; + test_recall_missing_scalar_measurement_impl(measurement, client) .await?; Ok(()) } async fn recall_measurement_u32_test(client: &Client) -> Result<(), Error> { let datum = Datum::U32(1); - let as_json = serde_json::Value::from(1_u32); - test_recall_measurement_impl::(datum, None, as_json, client) + let measurement = Measurement::new(Utc::now(), datum); + test_recall_measurement_impl(measurement.clone(), client).await?; + test_recall_missing_scalar_measurement_impl(measurement, client) .await?; Ok(()) } async fn recall_measurement_i64_test(client: &Client) -> Result<(), Error> { let datum = Datum::I64(1); - let as_json = serde_json::Value::from(1_i64); - test_recall_measurement_impl::(datum, None, as_json, client) + let measurement = Measurement::new(Utc::now(), datum); + test_recall_measurement_impl(measurement.clone(), client).await?; + test_recall_missing_scalar_measurement_impl(measurement, client) .await?; Ok(()) } async fn recall_measurement_u64_test(client: &Client) -> Result<(), Error> { let datum = Datum::U64(1); - let as_json = serde_json::Value::from(1_u64); - test_recall_measurement_impl::(datum, None, as_json, client) + let measurement = Measurement::new(Utc::now(), datum); + test_recall_measurement_impl(measurement.clone(), client).await?; + test_recall_missing_scalar_measurement_impl(measurement, client) .await?; Ok(()) } @@ -3034,9 +3062,9 @@ mod tests { async fn recall_measurement_f32_test(client: &Client) -> Result<(), Error> { const VALUE: f32 = 1.1; let datum = Datum::F32(VALUE); - // NOTE: This is intentionally an f64. - let as_json = serde_json::Value::from(1.1_f64); - test_recall_measurement_impl::(datum, None, as_json, client) + let measurement = Measurement::new(Utc::now(), datum); + test_recall_measurement_impl(measurement.clone(), client).await?; + test_recall_missing_scalar_measurement_impl(measurement, client) .await?; Ok(()) } @@ -3044,18 +3072,43 @@ mod tests { async fn recall_measurement_f64_test(client: &Client) -> Result<(), Error> { const VALUE: f64 = 1.1; let datum = Datum::F64(VALUE); - let as_json = serde_json::Value::from(VALUE); - test_recall_measurement_impl::(datum, None, as_json, client) + let measurement = Measurement::new(Utc::now(), datum); + test_recall_measurement_impl(measurement.clone(), client).await?; + test_recall_missing_scalar_measurement_impl(measurement, client) .await?; Ok(()) } + async fn recall_measurement_string_test( + client: &Client, + ) -> Result<(), Error> { + let value = String::from("foo"); + let datum = Datum::String(value.clone()); + let measurement = Measurement::new(Utc::now(), datum); + test_recall_measurement_impl(measurement.clone(), client).await?; + test_recall_missing_scalar_measurement_impl(measurement, client) + .await?; + Ok(()) + } + + async fn recall_measurement_bytes_test( + client: &Client, + ) -> Result<(), Error> { + let value = Bytes::from(vec![0, 1, 2]); + let datum = Datum::Bytes(value.clone()); + let measurement = Measurement::new(Utc::now(), datum); + test_recall_measurement_impl(measurement.clone(), client).await?; + // NOTE: We don't currently support missing byte array samples. + Ok(()) + } + async fn recall_measurement_cumulative_i64_test( client: &Client, ) -> Result<(), Error> { let datum = Datum::CumulativeI64(1.into()); - let as_json = serde_json::Value::from(1_i64); - test_recall_measurement_impl::(datum, None, as_json, client) + let measurement = Measurement::new(Utc::now(), datum); + test_recall_measurement_impl(measurement.clone(), client).await?; + test_recall_missing_scalar_measurement_impl(measurement, client) .await?; Ok(()) } @@ -3064,8 +3117,9 @@ mod tests { client: &Client, ) -> Result<(), Error> { let datum = Datum::CumulativeU64(1.into()); - let as_json = serde_json::Value::from(1_u64); - test_recall_measurement_impl::(datum, None, as_json, client) + let measurement = Measurement::new(Utc::now(), datum); + test_recall_measurement_impl(measurement.clone(), client).await?; + test_recall_missing_scalar_measurement_impl(measurement, client) .await?; Ok(()) } @@ -3074,8 +3128,9 @@ mod tests { client: &Client, ) -> Result<(), Error> { let datum = Datum::CumulativeF64(1.1.into()); - let as_json = serde_json::Value::from(1.1_f64); - test_recall_measurement_impl::(datum, None, as_json, client) + let measurement = Measurement::new(Utc::now(), datum); + test_recall_measurement_impl(measurement.clone(), client).await?; + test_recall_missing_scalar_measurement_impl(measurement, client) .await?; Ok(()) } @@ -3089,13 +3144,15 @@ mod tests { Datum: From>, serde_json::Value: From, { - let (bins, counts) = hist.to_arrays(); let datum = Datum::from(hist); - let as_json = serde_json::Value::Array( - counts.into_iter().map(Into::into).collect(), + let measurement = Measurement::new(Utc::now(), datum); + let missing_datum = Datum::Missing( + MissingDatum::new(measurement.datum_type(), Some(Utc::now())) + .unwrap(), ); - test_recall_measurement_impl(datum, Some(bins), as_json, client) - .await?; + let missing_measurement = Measurement::new(Utc::now(), missing_datum); + test_recall_measurement_impl(measurement, client).await?; + test_recall_measurement_impl(missing_measurement, client).await?; Ok(()) } @@ -3192,54 +3249,23 @@ mod tests { Ok(()) } - async fn test_recall_measurement_impl + Copy>( - datum: Datum, - maybe_bins: Option>, - json_datum: serde_json::Value, + async fn test_recall_measurement_impl( + measurement: Measurement, client: &Client, ) -> Result<(), Error> { // Insert a record from this datum. const TIMESERIES_NAME: &str = "foo:bar"; const TIMESERIES_KEY: u64 = 101; - let mut inserted_row = serde_json::Map::new(); - inserted_row - .insert("timeseries_name".to_string(), TIMESERIES_NAME.into()); - inserted_row - .insert("timeseries_key".to_string(), TIMESERIES_KEY.into()); - inserted_row.insert( - "timestamp".to_string(), - Utc::now() - .format(crate::DATABASE_TIMESTAMP_FORMAT) - .to_string() - .into(), - ); - - // Insert the start time and possibly bins. - if let Some(start_time) = datum.start_time() { - inserted_row.insert( - "start_time".to_string(), - start_time - .format(crate::DATABASE_TIMESTAMP_FORMAT) - .to_string() - .into(), - ); - } - if let Some(bins) = &maybe_bins { - let bins = serde_json::Value::Array( - bins.iter().copied().map(Into::into).collect(), + let (measurement_table, inserted_row) = + crate::model::unroll_measurement_row_impl( + TIMESERIES_NAME.to_string(), + TIMESERIES_KEY, + &measurement, ); - inserted_row.insert("bins".to_string(), bins); - inserted_row.insert("counts".to_string(), json_datum); - } else { - inserted_row.insert("datum".to_string(), json_datum); - } - let inserted_row = serde_json::Value::from(inserted_row); - - let measurement_table = measurement_table_name(datum.datum_type()); - let row = serde_json::to_string(&inserted_row).unwrap(); let insert_sql = format!( - "INSERT INTO oximeter.{measurement_table} FORMAT JSONEachRow {row}", + "INSERT INTO {measurement_table} FORMAT JSONEachRow {inserted_row}", ); + println!("Inserted row: {}", inserted_row); client .execute(insert_sql) .await @@ -3247,21 +3273,22 @@ mod tests { // Select it exactly back out. let select_sql = format!( - "SELECT * FROM oximeter.{} LIMIT 2 FORMAT {};", + "SELECT * FROM {} WHERE timestamp = '{}' FORMAT {};", measurement_table, + measurement.timestamp().format(crate::DATABASE_TIMESTAMP_FORMAT), crate::DATABASE_SELECT_FORMAT, ); let body = client .execute_with_body(select_sql) .await .expect("Failed to select measurement row"); - println!("{}", body); - let actual_row: serde_json::Value = serde_json::from_str(&body) - .expect("Failed to parse measurement row JSON"); - println!("{actual_row:?}"); - println!("{inserted_row:?}"); + let (_, actual_row) = crate::model::parse_measurement_from_row( + &body, + measurement.datum_type(), + ); + println!("Actual row: {actual_row:?}"); assert_eq!( - actual_row, inserted_row, + actual_row, measurement, "Actual and expected measurement rows do not match" ); Ok(()) @@ -3311,6 +3338,10 @@ mod tests { recall_measurement_f64_test(&client).await.unwrap(); + recall_measurement_string_test(&client).await.unwrap(); + + recall_measurement_bytes_test(&client).await.unwrap(); + recall_measurement_cumulative_i64_test(&client).await.unwrap(); recall_measurement_cumulative_u64_test(&client).await.unwrap(); diff --git a/oximeter/db/src/model.rs b/oximeter/db/src/model.rs index 715e025a04d..fc3b8a17c94 100644 --- a/oximeter/db/src/model.rs +++ b/oximeter/db/src/model.rs @@ -26,6 +26,7 @@ use oximeter::types::Field; use oximeter::types::FieldType; use oximeter::types::FieldValue; use oximeter::types::Measurement; +use oximeter::types::MissingDatum; use oximeter::types::Sample; use serde::Deserialize; use serde::Serialize; @@ -43,7 +44,7 @@ use uuid::Uuid; /// - [`crate::Client::initialize_db_with_version`] /// - [`crate::Client::ensure_schema`] /// - The `clickhouse-schema-updater` binary in this crate -pub const OXIMETER_VERSION: u64 = 3; +pub const OXIMETER_VERSION: u64 = 4; // Wrapper type to represent a boolean in the database. // @@ -212,6 +213,7 @@ impl From for DbFieldType { } } } + #[derive(Clone, Copy, Debug, Serialize, Deserialize, PartialEq)] pub enum DbDatumType { Bool, @@ -402,7 +404,7 @@ macro_rules! declare_measurement_row { timeseries_key: TimeseriesKey, #[serde(with = "serde_timestamp")] timestamp: DateTime, - datum: $datum_type, + datum: Option<$datum_type>, } impl_table_name!{$name, "measurements", $data_type} @@ -433,7 +435,7 @@ macro_rules! declare_cumulative_measurement_row { start_time: DateTime, #[serde(with = "serde_timestamp")] timestamp: DateTime, - datum: $datum_type, + datum: Option<$datum_type>, } impl_table_name!{$name, "measurements", $data_type} @@ -456,6 +458,22 @@ struct DbHistogram { pub counts: Vec, } +// We use an empty histogram to indicate a missing sample. +// +// While ClickHouse supports nullable types, the inner type can't be a +// "composite", which includes arrays. I.e., `Nullable(Array(UInt8))` can't be +// used. This is unfortunate, but we are aided by the fact that it's not +// possible to have an `oximeter` histogram that contains zero bins right now. +// This is checked by a test in `oximeter::histogram`. +// +// That means we can currently use an empty array from the database as a +// sentinel for a missing sample. +impl DbHistogram { + fn null() -> Self { + Self { bins: vec![], counts: vec![] } + } +} + impl From<&Histogram> for DbHistogram where T: traits::HistogramSupport, @@ -647,267 +665,412 @@ pub(crate) fn unroll_measurement_row(sample: &Sample) -> (String, String) { let timeseries_name = sample.timeseries_name.clone(); let timeseries_key = crate::timeseries_key(sample); let measurement = &sample.measurement; + unroll_measurement_row_impl(timeseries_name, timeseries_key, measurement) +} + +pub(crate) fn unroll_measurement_row_impl( + timeseries_name: String, + timeseries_key: TimeseriesKey, + measurement: &Measurement, +) -> (String, String) { let timestamp = measurement.timestamp(); let extract_start_time = |measurement: &Measurement| { measurement .start_time() .expect("Cumulative measurements must have a start time") }; - match measurement.datum() { - Datum::Bool(inner) => { + + let datum = measurement.datum(); + match measurement.datum_type() { + DatumType::Bool => { + let datum = match datum { + Datum::Bool(inner) => Some(DbBool::from(*inner)), + Datum::Missing(_) => None, + _ => unreachable!(), + }; let row = BoolMeasurementRow { timeseries_name, timeseries_key, timestamp, - datum: DbBool::from(*inner), + datum, }; (row.table_name(), serde_json::to_string(&row).unwrap()) } - Datum::I8(inner) => { + DatumType::I8 => { + let datum = match datum { + Datum::I8(inner) => Some(*inner), + Datum::Missing(_) => None, + _ => unreachable!(), + }; let row = I8MeasurementRow { timeseries_name, timeseries_key, timestamp, - datum: *inner, + datum, }; (row.table_name(), serde_json::to_string(&row).unwrap()) } - Datum::U8(inner) => { + DatumType::U8 => { + let datum = match datum { + Datum::U8(inner) => Some(*inner), + Datum::Missing(_) => None, + _ => unreachable!(), + }; let row = U8MeasurementRow { timeseries_name, timeseries_key, timestamp, - datum: *inner, + datum, }; (row.table_name(), serde_json::to_string(&row).unwrap()) } - Datum::I16(inner) => { + DatumType::I16 => { + let datum = match datum { + Datum::I16(inner) => Some(*inner), + Datum::Missing(_) => None, + _ => unreachable!(), + }; let row = I16MeasurementRow { timeseries_name, timeseries_key, timestamp, - datum: *inner, + datum, }; (row.table_name(), serde_json::to_string(&row).unwrap()) } - Datum::U16(inner) => { + DatumType::U16 => { + let datum = match datum { + Datum::U16(inner) => Some(*inner), + Datum::Missing(_) => None, + _ => unreachable!(), + }; let row = U16MeasurementRow { timeseries_name, timeseries_key, timestamp, - datum: *inner, + datum, }; (row.table_name(), serde_json::to_string(&row).unwrap()) } - Datum::I32(inner) => { + DatumType::I32 => { + let datum = match datum { + Datum::I32(inner) => Some(*inner), + Datum::Missing(_) => None, + _ => unreachable!(), + }; let row = I32MeasurementRow { timeseries_name, timeseries_key, timestamp, - datum: *inner, + datum, }; (row.table_name(), serde_json::to_string(&row).unwrap()) } - Datum::U32(inner) => { + DatumType::U32 => { + let datum = match datum { + Datum::U32(inner) => Some(*inner), + Datum::Missing(_) => None, + _ => unreachable!(), + }; let row = U32MeasurementRow { timeseries_name, timeseries_key, timestamp, - datum: *inner, + datum, }; (row.table_name(), serde_json::to_string(&row).unwrap()) } - Datum::I64(inner) => { + DatumType::I64 => { + let datum = match datum { + Datum::I64(inner) => Some(*inner), + Datum::Missing(_) => None, + _ => unreachable!(), + }; let row = I64MeasurementRow { timeseries_name, timeseries_key, timestamp, - datum: *inner, + datum, }; (row.table_name(), serde_json::to_string(&row).unwrap()) } - Datum::U64(inner) => { + DatumType::U64 => { + let datum = match datum { + Datum::U64(inner) => Some(*inner), + Datum::Missing(_) => None, + _ => unreachable!(), + }; let row = U64MeasurementRow { timeseries_name, timeseries_key, timestamp, - datum: *inner, + datum, }; (row.table_name(), serde_json::to_string(&row).unwrap()) } - Datum::F32(inner) => { + DatumType::F32 => { + let datum = match datum { + Datum::F32(inner) => Some(*inner), + Datum::Missing(_) => None, + _ => unreachable!(), + }; let row = F32MeasurementRow { timeseries_name, timeseries_key, timestamp, - datum: *inner, + datum, }; (row.table_name(), serde_json::to_string(&row).unwrap()) } - Datum::F64(inner) => { + DatumType::F64 => { + let datum = match datum { + Datum::F64(inner) => Some(*inner), + Datum::Missing(_) => None, + _ => unreachable!(), + }; let row = F64MeasurementRow { timeseries_name, timeseries_key, timestamp, - datum: *inner, + datum, }; (row.table_name(), serde_json::to_string(&row).unwrap()) } - Datum::String(ref inner) => { + DatumType::String => { + let datum = match datum { + Datum::String(ref inner) => Some(inner.clone()), + Datum::Missing(_) => None, + _ => unreachable!(), + }; let row = StringMeasurementRow { timeseries_name, timeseries_key, timestamp, - datum: inner.clone(), + datum, }; (row.table_name(), serde_json::to_string(&row).unwrap()) } - Datum::Bytes(ref inner) => { + DatumType::Bytes => { + let datum = match datum { + Datum::Bytes(ref inner) => Some(inner.clone()), + Datum::Missing(_) => None, + _ => unreachable!(), + }; let row = BytesMeasurementRow { timeseries_name, timeseries_key, timestamp, - datum: inner.clone(), + datum, }; (row.table_name(), serde_json::to_string(&row).unwrap()) } - Datum::CumulativeI64(inner) => { + DatumType::CumulativeI64 => { + let datum = match datum { + Datum::CumulativeI64(inner) => Some(inner.value()), + Datum::Missing(_) => None, + _ => unreachable!(), + }; let row = CumulativeI64MeasurementRow { timeseries_name, timeseries_key, start_time: extract_start_time(measurement), timestamp, - datum: inner.value(), + datum, }; (row.table_name(), serde_json::to_string(&row).unwrap()) } - Datum::CumulativeU64(inner) => { + DatumType::CumulativeU64 => { + let datum = match datum { + Datum::CumulativeU64(inner) => Some(inner.value()), + Datum::Missing(_) => None, + _ => unreachable!(), + }; let row = CumulativeU64MeasurementRow { timeseries_name, timeseries_key, start_time: extract_start_time(measurement), timestamp, - datum: inner.value(), + datum, }; (row.table_name(), serde_json::to_string(&row).unwrap()) } - Datum::CumulativeF32(inner) => { + DatumType::CumulativeF32 => { + let datum = match datum { + Datum::CumulativeF32(inner) => Some(inner.value()), + Datum::Missing(_) => None, + _ => unreachable!(), + }; let row = CumulativeF32MeasurementRow { timeseries_name, timeseries_key, start_time: extract_start_time(measurement), timestamp, - datum: inner.value(), + datum, }; (row.table_name(), serde_json::to_string(&row).unwrap()) } - Datum::CumulativeF64(inner) => { + DatumType::CumulativeF64 => { + let datum = match datum { + Datum::CumulativeF64(inner) => Some(inner.value()), + Datum::Missing(_) => None, + _ => unreachable!(), + }; let row = CumulativeF64MeasurementRow { timeseries_name, timeseries_key, start_time: extract_start_time(measurement), timestamp, - datum: inner.value(), + datum, }; (row.table_name(), serde_json::to_string(&row).unwrap()) } - Datum::HistogramI8(ref inner) => { + DatumType::HistogramI8 => { + let datum = match datum { + Datum::HistogramI8(inner) => DbHistogram::from(inner), + Datum::Missing(_) => DbHistogram::null(), + _ => unreachable!(), + }; let row = HistogramI8MeasurementRow { timeseries_name, timeseries_key, start_time: extract_start_time(measurement), timestamp, - datum: DbHistogram::from(inner), + datum, }; (row.table_name(), serde_json::to_string(&row).unwrap()) } - Datum::HistogramU8(ref inner) => { + DatumType::HistogramU8 => { + let datum = match datum { + Datum::HistogramU8(inner) => DbHistogram::from(inner), + Datum::Missing(_) => DbHistogram::null(), + _ => unreachable!(), + }; let row = HistogramU8MeasurementRow { timeseries_name, timeseries_key, start_time: extract_start_time(measurement), timestamp, - datum: DbHistogram::from(inner), + datum, }; (row.table_name(), serde_json::to_string(&row).unwrap()) } - Datum::HistogramI16(ref inner) => { + DatumType::HistogramI16 => { + let datum = match datum { + Datum::HistogramI16(inner) => DbHistogram::from(inner), + Datum::Missing(_) => DbHistogram::null(), + _ => unreachable!(), + }; let row = HistogramI16MeasurementRow { timeseries_name, timeseries_key, start_time: extract_start_time(measurement), timestamp, - datum: DbHistogram::from(inner), + datum, }; (row.table_name(), serde_json::to_string(&row).unwrap()) } - Datum::HistogramU16(ref inner) => { + DatumType::HistogramU16 => { + let datum = match datum { + Datum::HistogramU16(inner) => DbHistogram::from(inner), + Datum::Missing(_) => DbHistogram::null(), + _ => unreachable!(), + }; let row = HistogramU16MeasurementRow { timeseries_name, timeseries_key, start_time: extract_start_time(measurement), timestamp, - datum: DbHistogram::from(inner), + datum, }; (row.table_name(), serde_json::to_string(&row).unwrap()) } - Datum::HistogramI32(ref inner) => { + DatumType::HistogramI32 => { + let datum = match datum { + Datum::HistogramI32(inner) => DbHistogram::from(inner), + Datum::Missing(_) => DbHistogram::null(), + _ => unreachable!(), + }; let row = HistogramI32MeasurementRow { timeseries_name, timeseries_key, start_time: extract_start_time(measurement), timestamp, - datum: DbHistogram::from(inner), + datum, }; (row.table_name(), serde_json::to_string(&row).unwrap()) } - Datum::HistogramU32(ref inner) => { + DatumType::HistogramU32 => { + let datum = match datum { + Datum::HistogramU32(inner) => DbHistogram::from(inner), + Datum::Missing(_) => DbHistogram::null(), + _ => unreachable!(), + }; let row = HistogramU32MeasurementRow { timeseries_name, timeseries_key, start_time: extract_start_time(measurement), timestamp, - datum: DbHistogram::from(inner), + datum, }; (row.table_name(), serde_json::to_string(&row).unwrap()) } - Datum::HistogramI64(ref inner) => { + DatumType::HistogramI64 => { + let datum = match datum { + Datum::HistogramI64(inner) => DbHistogram::from(inner), + Datum::Missing(_) => DbHistogram::null(), + _ => unreachable!(), + }; let row = HistogramI64MeasurementRow { timeseries_name, timeseries_key, start_time: extract_start_time(measurement), timestamp, - datum: DbHistogram::from(inner), + datum, }; (row.table_name(), serde_json::to_string(&row).unwrap()) } - Datum::HistogramU64(ref inner) => { + DatumType::HistogramU64 => { + let datum = match datum { + Datum::HistogramU64(inner) => DbHistogram::from(inner), + Datum::Missing(_) => DbHistogram::null(), + _ => unreachable!(), + }; let row = HistogramU64MeasurementRow { timeseries_name, timeseries_key, start_time: extract_start_time(measurement), timestamp, - datum: DbHistogram::from(inner), + datum, }; (row.table_name(), serde_json::to_string(&row).unwrap()) } - Datum::HistogramF32(ref inner) => { + DatumType::HistogramF32 => { + let datum = match datum { + Datum::HistogramF32(inner) => DbHistogram::from(inner), + Datum::Missing(_) => DbHistogram::null(), + _ => unreachable!(), + }; let row = HistogramF32MeasurementRow { timeseries_name, timeseries_key, start_time: extract_start_time(measurement), timestamp, - datum: DbHistogram::from(inner), + datum, }; (row.table_name(), serde_json::to_string(&row).unwrap()) } - Datum::HistogramF64(ref inner) => { + DatumType::HistogramF64 => { + let datum = match datum { + Datum::HistogramF64(inner) => DbHistogram::from(inner), + Datum::Missing(_) => DbHistogram::null(), + _ => unreachable!(), + }; let row = HistogramF64MeasurementRow { timeseries_name, timeseries_key, start_time: extract_start_time(measurement), timestamp, - datum: DbHistogram::from(inner), + datum, }; (row.table_name(), serde_json::to_string(&row).unwrap()) } @@ -984,7 +1147,7 @@ struct DbTimeseriesScalarGaugeSample { timeseries_key: TimeseriesKey, #[serde(with = "serde_timestamp")] timestamp: DateTime, - datum: T, + datum: Option, } // A scalar timestamped sample from a cumulative timeseries, as extracted from a query to the @@ -996,7 +1159,7 @@ struct DbTimeseriesScalarCumulativeSample { start_time: DateTime, #[serde(with = "serde_timestamp")] timestamp: DateTime, - datum: T, + datum: Option, } // A histogram timestamped sample from a timeseries, as extracted from a query to the database. @@ -1014,9 +1177,15 @@ struct DbTimeseriesHistogramSample { impl From> for Measurement where Datum: From, + T: FromDbScalar, { fn from(sample: DbTimeseriesScalarGaugeSample) -> Measurement { - let datum = Datum::from(sample.datum); + let datum = match sample.datum { + Some(datum) => Datum::from(datum), + None => { + Datum::Missing(MissingDatum::new(T::DATUM_TYPE, None).unwrap()) + } + }; Measurement::new(sample.timestamp, datum) } } @@ -1024,12 +1193,19 @@ where impl From> for Measurement where Datum: From>, - T: traits::Cumulative, + T: traits::Cumulative + FromDbCumulative, { fn from(sample: DbTimeseriesScalarCumulativeSample) -> Measurement { - let cumulative = - Cumulative::with_start_time(sample.start_time, sample.datum); - let datum = Datum::from(cumulative); + let datum = match sample.datum { + Some(datum) => Datum::from(Cumulative::with_start_time( + sample.start_time, + datum, + )), + None => Datum::Missing( + MissingDatum::new(T::DATUM_TYPE, Some(sample.start_time)) + .unwrap(), + ), + }; Measurement::new(sample.timestamp, datum) } } @@ -1037,26 +1213,157 @@ where impl From> for Measurement where Datum: From>, - T: traits::HistogramSupport, + T: traits::HistogramSupport + FromDbHistogram, { fn from(sample: DbTimeseriesHistogramSample) -> Measurement { - let datum = Datum::from( - Histogram::from_arrays( - sample.start_time, - sample.bins, - sample.counts, + let datum = if sample.bins.is_empty() { + assert!(sample.counts.is_empty()); + Datum::Missing( + MissingDatum::new(T::DATUM_TYPE, Some(sample.start_time)) + .unwrap(), ) - .unwrap(), - ); + } else { + Datum::from( + Histogram::from_arrays( + sample.start_time, + sample.bins, + sample.counts, + ) + .unwrap(), + ) + }; Measurement::new(sample.timestamp, datum) } } +// Helper trait providing the DatumType for a corresponding scalar DB value. +// +// This is used in `parse_timeseries_scalar_gauge_measurement`. +trait FromDbScalar { + const DATUM_TYPE: DatumType; +} + +impl FromDbScalar for DbBool { + const DATUM_TYPE: DatumType = DatumType::Bool; +} + +impl FromDbScalar for i8 { + const DATUM_TYPE: DatumType = DatumType::I8; +} + +impl FromDbScalar for u8 { + const DATUM_TYPE: DatumType = DatumType::U8; +} + +impl FromDbScalar for i16 { + const DATUM_TYPE: DatumType = DatumType::I16; +} + +impl FromDbScalar for u16 { + const DATUM_TYPE: DatumType = DatumType::U16; +} + +impl FromDbScalar for i32 { + const DATUM_TYPE: DatumType = DatumType::I32; +} + +impl FromDbScalar for u32 { + const DATUM_TYPE: DatumType = DatumType::U32; +} + +impl FromDbScalar for i64 { + const DATUM_TYPE: DatumType = DatumType::I64; +} + +impl FromDbScalar for u64 { + const DATUM_TYPE: DatumType = DatumType::U64; +} + +impl FromDbScalar for f32 { + const DATUM_TYPE: DatumType = DatumType::F32; +} + +impl FromDbScalar for f64 { + const DATUM_TYPE: DatumType = DatumType::F64; +} + +impl FromDbScalar for String { + const DATUM_TYPE: DatumType = DatumType::String; +} + +impl FromDbScalar for Bytes { + const DATUM_TYPE: DatumType = DatumType::Bytes; +} + +trait FromDbCumulative { + const DATUM_TYPE: DatumType; +} + +impl FromDbCumulative for i64 { + const DATUM_TYPE: DatumType = DatumType::CumulativeI64; +} + +impl FromDbCumulative for u64 { + const DATUM_TYPE: DatumType = DatumType::CumulativeU64; +} + +impl FromDbCumulative for f32 { + const DATUM_TYPE: DatumType = DatumType::CumulativeF32; +} + +impl FromDbCumulative for f64 { + const DATUM_TYPE: DatumType = DatumType::CumulativeF64; +} + +trait FromDbHistogram { + const DATUM_TYPE: DatumType; +} + +impl FromDbHistogram for i8 { + const DATUM_TYPE: DatumType = DatumType::HistogramI8; +} + +impl FromDbHistogram for u8 { + const DATUM_TYPE: DatumType = DatumType::HistogramU8; +} + +impl FromDbHistogram for i16 { + const DATUM_TYPE: DatumType = DatumType::HistogramI16; +} + +impl FromDbHistogram for u16 { + const DATUM_TYPE: DatumType = DatumType::HistogramU16; +} + +impl FromDbHistogram for i32 { + const DATUM_TYPE: DatumType = DatumType::HistogramI32; +} + +impl FromDbHistogram for u32 { + const DATUM_TYPE: DatumType = DatumType::HistogramU32; +} + +impl FromDbHistogram for i64 { + const DATUM_TYPE: DatumType = DatumType::HistogramI64; +} + +impl FromDbHistogram for u64 { + const DATUM_TYPE: DatumType = DatumType::HistogramU64; +} + +impl FromDbHistogram for f32 { + const DATUM_TYPE: DatumType = DatumType::HistogramF32; +} + +impl FromDbHistogram for f64 { + const DATUM_TYPE: DatumType = DatumType::HistogramF64; +} + fn parse_timeseries_scalar_gauge_measurement<'a, T>( line: &'a str, ) -> (TimeseriesKey, Measurement) where - T: Deserialize<'a> + Into, + T: Deserialize<'a> + Into + FromDbScalar, Datum: From, { let sample = @@ -1068,7 +1375,7 @@ fn parse_timeseries_scalar_cumulative_measurement<'a, T>( line: &'a str, ) -> (TimeseriesKey, Measurement) where - T: Deserialize<'a> + traits::Cumulative, + T: Deserialize<'a> + traits::Cumulative + FromDbCumulative, Datum: From>, { let sample = @@ -1081,7 +1388,7 @@ fn parse_timeseries_histogram_measurement( line: &str, ) -> (TimeseriesKey, Measurement) where - T: Into + traits::HistogramSupport, + T: Into + traits::HistogramSupport + FromDbHistogram, Datum: From>, { let sample = @@ -1459,6 +1766,27 @@ mod tests { } } + // Test that we correctly unroll a row when the measurement is missing its + // datum. + #[test] + fn test_unroll_missing_measurement_row() { + let sample = test_util::make_sample(); + let missing_sample = test_util::make_missing_sample(); + let (table_name, row) = unroll_measurement_row(&sample); + let (missing_table_name, missing_row) = + unroll_measurement_row(&missing_sample); + let row = serde_json::from_str::(&row).unwrap(); + let missing_row = + serde_json::from_str::(&missing_row).unwrap(); + println!("{row:#?}"); + println!("{missing_row:#?}"); + assert_eq!(table_name, missing_table_name); + assert_eq!(row.timeseries_name, missing_row.timeseries_name); + assert_eq!(row.timeseries_key, missing_row.timeseries_key); + assert!(row.datum.is_some()); + assert!(missing_row.datum.is_none()); + } + #[test] fn test_unroll_measurement_row() { let sample = test_util::make_hist_sample(); @@ -1473,14 +1801,13 @@ mod tests { ) .unwrap(); let measurement = &sample.measurement; - if let Datum::HistogramF64(hist) = measurement.datum() { - assert_eq!( - hist, &unpacked_hist, - "Unpacking histogram from database representation failed" - ); - } else { + let Datum::HistogramF64(hist) = measurement.datum() else { panic!("Expected a histogram measurement"); - } + }; + assert_eq!( + hist, &unpacked_hist, + "Unpacking histogram from database representation failed" + ); assert_eq!(unpacked.start_time, measurement.start_time().unwrap()); } @@ -1582,12 +1909,11 @@ mod tests { assert_eq!(key, 12); assert_eq!(measurement.start_time().unwrap(), start_time); assert_eq!(measurement.timestamp(), timestamp); - if let Datum::HistogramI64(hist) = measurement.datum() { - assert_eq!(hist.n_bins(), 3); - assert_eq!(hist.n_samples(), 2); - } else { + let Datum::HistogramI64(hist) = measurement.datum() else { panic!("Expected a histogram sample"); - } + }; + assert_eq!(hist.n_bins(), 3); + assert_eq!(hist.n_samples(), 2); } #[test] @@ -1624,4 +1950,14 @@ mod tests { "Histogram reconstructed from paired arrays is not correct" ); } + #[test] + fn test_parse_bytes_measurement() { + let s = r#"{"timeseries_key": 101, "timestamp": "2023-11-21 18:25:21.963714255", "datum": "\u0001\u0002\u0003"}"#; + let (_, meas) = parse_timeseries_scalar_gauge_measurement::(&s); + println!("{meas:?}"); + let Datum::Bytes(b) = meas.datum() else { + unreachable!(); + }; + assert_eq!(b.to_vec(), vec![1, 2, 3]); + } } diff --git a/oximeter/oximeter/Cargo.toml b/oximeter/oximeter/Cargo.toml index 8a69494d5ae..0cb2d8cace8 100644 --- a/oximeter/oximeter/Cargo.toml +++ b/oximeter/oximeter/Cargo.toml @@ -21,4 +21,5 @@ omicron-workspace-hack.workspace = true [dev-dependencies] approx.workspace = true rstest.workspace = true +serde_json.workspace = true trybuild.workspace = true diff --git a/oximeter/oximeter/src/histogram.rs b/oximeter/oximeter/src/histogram.rs index c399384ffaf..aaf9297ca4e 100644 --- a/oximeter/oximeter/src/histogram.rs +++ b/oximeter/oximeter/src/histogram.rs @@ -1353,13 +1353,10 @@ mod tests { } #[test] - fn test_foo() { - let bins: Vec = 10u16.bins(1, 3, 30.try_into().unwrap()).unwrap(); - println!("{bins:?}"); - dbg!(bins.len()); - let hist = Histogram::new(&bins).unwrap(); - for bin in hist.iter() { - println!("{}", bin.range); - } + fn test_empty_bins_not_supported() { + assert!(matches!( + Histogram::::new(&[]).unwrap_err(), + HistogramError::EmptyBins + )); } } diff --git a/oximeter/oximeter/src/test_util.rs b/oximeter/oximeter/src/test_util.rs index f3750d6d839..a9778d03bc2 100644 --- a/oximeter/oximeter/src/test_util.rs +++ b/oximeter/oximeter/src/test_util.rs @@ -48,19 +48,27 @@ pub struct TestHistogram { pub datum: Histogram, } +const ID: Uuid = uuid::uuid!("e00ced4d-39d1-446a-ae85-a67f05c9750b"); + pub fn make_sample() -> Sample { let target = TestTarget::default(); - let metric = TestMetric { id: Uuid::new_v4(), good: true, datum: 1 }; + let metric = TestMetric { id: ID, good: true, datum: 1 }; Sample::new(&target, &metric).unwrap() } +pub fn make_missing_sample() -> Sample { + let target = TestTarget::default(); + let metric = TestMetric { id: ID, good: true, datum: 1 }; + Sample::new_missing(&target, &metric).unwrap() +} + pub fn make_hist_sample() -> Sample { let target = TestTarget::default(); let mut hist = histogram::Histogram::new(&[0.0, 5.0, 10.0]).unwrap(); hist.sample(1.0).unwrap(); hist.sample(2.0).unwrap(); hist.sample(6.0).unwrap(); - let metric = TestHistogram { id: Uuid::new_v4(), good: true, datum: hist }; + let metric = TestHistogram { id: ID, good: true, datum: hist }; Sample::new(&target, &metric).unwrap() } diff --git a/oximeter/oximeter/src/traits.rs b/oximeter/oximeter/src/traits.rs index 096abb80233..0934d231e38 100644 --- a/oximeter/oximeter/src/traits.rs +++ b/oximeter/oximeter/src/traits.rs @@ -30,8 +30,15 @@ use std::ops::AddAssign; /// definition can be thought of as a schema, and an instance of that struct as identifying an /// individual target. /// -/// Target fields may have one of a set of supported types: `bool`, `i64`, `String`, `IpAddr`, or -/// `Uuid`. Any number of fields greater than zero is supported. +/// Target fields may have one of a set of supported types: +/// +/// - `bool` +/// - any fixed-width integer, e.g., `u8` or `i64` +/// - `String` +/// - `IpAddr` +/// - `Uuid` +/// +/// Any number of fields greater than zero is supported. /// /// Examples /// -------- @@ -105,9 +112,28 @@ pub trait Target { /// One field of the struct is special, describing the actual measured data that the metric /// represents. This should be a field named `datum`, or another field (with any name you choose) /// annotated with the `#[datum]` attribute. This field represents the underlying data for the -/// metric, and must be one of the supported types, implementing the [`Datum`] trait. This can -/// be any of: `i64`, `f64`, `bool`, `String`, or `Bytes` for gauges, and `Cumulative` or -/// `Histogram` for cumulative metrics, where `T` is `i64` or `f64`. +/// metric, and must be one of the supported types, implementing the [`Datum`] trait. +/// +/// For gauge types, this can be any of: +/// +/// - `bool` +/// - a fixed-width integer, e.g. `u8` or `i64` +/// - `f32` or `f64` +/// - `String` +/// - `Bytes` +/// +/// Cumulative types can be any of `Cumulative`, where `T` is +/// +/// - `i64` +/// - `u64` +/// - `f32` +/// - `f64` +/// +/// Histogram types can be any `Histogram`, wher `T` is: +/// +/// - a fixed-width integer, e.g. `u8` or `i64` +/// - `f32` +/// - `f64` /// /// The value of the metric's data is _measured_ by using the `measure()` method, which returns a /// [`Measurement`]. This describes a timestamped data point for the metric. diff --git a/oximeter/oximeter/src/types.rs b/oximeter/oximeter/src/types.rs index 325974781e2..ca80f1153ae 100644 --- a/oximeter/oximeter/src/types.rs +++ b/oximeter/oximeter/src/types.rs @@ -330,6 +330,23 @@ impl DatumType { | DatumType::HistogramF64 ) } + + /// Return `true` if this datum type is a histogram. + pub fn is_histogram(&self) -> bool { + matches!( + self, + DatumType::HistogramI8 + | DatumType::HistogramU8 + | DatumType::HistogramI16 + | DatumType::HistogramU16 + | DatumType::HistogramI32 + | DatumType::HistogramU32 + | DatumType::HistogramI64 + | DatumType::HistogramU64 + | DatumType::HistogramF32 + | DatumType::HistogramF64 + ) + } } impl std::fmt::Display for DatumType { @@ -369,6 +386,7 @@ pub enum Datum { HistogramU64(histogram::Histogram), HistogramF32(histogram::Histogram), HistogramF64(histogram::Histogram), + Missing(MissingDatum), } impl Datum { @@ -402,6 +420,7 @@ impl Datum { Datum::HistogramU64(_) => DatumType::HistogramU64, Datum::HistogramF32(_) => DatumType::HistogramF32, Datum::HistogramF64(_) => DatumType::HistogramF64, + Datum::Missing(ref inner) => inner.datum_type(), } } @@ -440,6 +459,7 @@ impl Datum { Datum::HistogramU64(ref inner) => Some(inner.start_time()), Datum::HistogramF32(ref inner) => Some(inner.start_time()), Datum::HistogramF64(ref inner) => Some(inner.start_time()), + Datum::Missing(ref inner) => inner.start_time(), } } } @@ -495,6 +515,60 @@ impl From<&str> for Datum { } } +#[derive(Clone, Copy, Debug, Deserialize, JsonSchema, PartialEq, Serialize)] +pub struct MissingDatum { + datum_type: DatumType, + start_time: Option>, +} + +impl MissingDatum { + pub fn datum_type(&self) -> DatumType { + self.datum_type + } + + pub fn start_time(&self) -> Option> { + self.start_time + } + + pub fn new( + datum_type: DatumType, + start_time: Option>, + ) -> Result { + // See https://github.com/oxidecomputer/omicron/issues/4551. + if datum_type == DatumType::Bytes { + return Err(MetricsError::DatumError(String::from( + "Missing samples from byte array types are not supported", + ))); + } + if datum_type.is_cumulative() && start_time.is_none() { + return Err(MetricsError::MissingDatumRequiresStartTime { + datum_type, + }); + } + if !datum_type.is_cumulative() && start_time.is_some() { + return Err(MetricsError::MissingDatumCannotHaveStartTime { + datum_type, + }); + } + Ok(Self { datum_type, start_time }) + } +} + +impl From for Datum { + fn from(d: MissingDatum) -> Datum { + Datum::Missing(d) + } +} + +impl From<&M> for MissingDatum { + fn from(metric: &M) -> Self { + MissingDatum { + datum_type: metric.datum_type(), + start_time: metric.start_time(), + } + } +} + /// A `Measurement` is a timestamped datum from a single metric #[derive(Clone, Debug, PartialEq, JsonSchema, Serialize, Deserialize)] pub struct Measurement { @@ -516,6 +590,11 @@ impl Measurement { Self { timestamp, datum: datum.into() } } + /// Return true if this measurement represents a missing datum. + pub fn is_missing(&self) -> bool { + matches!(self.datum, Datum::Missing(_)) + } + /// Return the datum for this measurement pub fn datum(&self) -> &Datum { &self.datum @@ -561,6 +640,12 @@ pub enum MetricsError { /// A field name is duplicated between the target and metric. #[error("Field '{name}' is duplicated between the target and metric")] DuplicateFieldName { name: String }, + + #[error("Missing datum of type {datum_type} requires a start time")] + MissingDatumRequiresStartTime { datum_type: DatumType }, + + #[error("Missing datum of type {datum_type} cannot have a start time")] + MissingDatumCannotHaveStartTime { datum_type: DatumType }, } impl From for omicron_common::api::external::Error { @@ -734,6 +819,29 @@ impl Sample { }) } + /// Construct a new missing sample, recorded at the time of the supplied + /// timestamp. + pub fn new_missing_with_timestamp( + timestamp: DateTime, + target: &T, + metric: &M, + ) -> Result + where + T: traits::Target, + M: traits::Metric, + { + let target_fields = FieldSet::from_target(target); + let metric_fields = FieldSet::from_metric(metric); + Self::verify_field_names(&target_fields, &metric_fields)?; + let datum = Datum::Missing(MissingDatum::from(metric)); + Ok(Self { + timeseries_name: crate::timeseries_name(target, metric), + target: target_fields, + metric: metric_fields, + measurement: Measurement { timestamp, datum }, + }) + } + /// Construct a new sample, created at the time the function is called. /// /// This materializes the data from the target and metric, and stores that information along @@ -746,6 +854,18 @@ impl Sample { Self::new_with_timestamp(Utc::now(), target, metric) } + /// Construct a new sample with a missing measurement. + pub fn new_missing( + target: &T, + metric: &M, + ) -> Result + where + T: traits::Target, + M: traits::Metric, + { + Self::new_missing_with_timestamp(Utc::now(), target, metric) + } + /// Return the fields for this sample. /// /// This returns the target fields and metric fields, chained, although there is no distinction @@ -951,7 +1071,7 @@ mod tests { fn test_measurement() { let measurement = Measurement::new(chrono::Utc::now(), 0i64); assert_eq!(measurement.datum_type(), DatumType::I64); - assert_eq!(measurement.start_time(), None); + assert!(measurement.start_time().is_none()); let datum = Cumulative::new(0i64); let measurement = Measurement::new(chrono::Utc::now(), datum); @@ -1032,4 +1152,11 @@ mod tests { Err(MetricsError::DuplicateFieldName { .. }) )); } + + #[test] + fn test_foo() { + // Datum + let s = r#"{"type": "cumulative_u64", "datum": { "start_time": "2023-11-17T22:42:17.299717021Z", "value": 0 } }"#; + println!("{:#?}", serde_json::from_str::(s).unwrap()); + } }