diff --git a/cmd/metrics-gen/TODO.md b/cmd/metrics-gen/TODO.md new file mode 100644 index 0000000..4416f2e --- /dev/null +++ b/cmd/metrics-gen/TODO.md @@ -0,0 +1,344 @@ +# Metrics Generator: typed error metrics + +Status: design handoff for a future session. The JSON metrics exist, but +successful operations currently export `error.type=""`. The work below has not +been implemented yet. + +## Objective + +Teach `metrics-gen` that errors are a special kind of metric attribute: + +- a successful operation must omit `error.type`; +- a failed operation must include one predictable, low-cardinality + `error.type` value; +- allowed error values should be declared once in the metrics specification; +- generated code should expose separate success and failure recording methods; +- known error attribute sets should be built once during metric initialization, + not reconstructed for every measurement. + +This follows the OpenTelemetry error convention. An empty string is a real +attribute value, not the same thing as an absent attribute. `_OTHER` is the +fallback for an unclassified failure; it must not represent success. + +## Current state + +- `processor/metrics/spec.yaml` declares `error.type` as a normal attribute set. +- Generated JSON duration methods therefore require `errType string` on every + call. +- `processor/json_encoder.go` and `processor/json_decoder.go` call + `getJSONErrorType(nil)`, receive `""`, and record `error.type=""` on success. +- `processor/json.go` already classifies JSON errors into stable strings. +- Duration bucket-set generation and float normalization are already + implemented. Do not undo the clean `1e-06, 2.5e-06, ...` output. + +## Recommended YAML model + +Use `errors` for an inline list and `error_set` for a named reusable set. Avoid +a singular `error` field: it sounds like one runtime error rather than a list +of allowed error types. + +```yaml +error_sets: + - name: json_decoder + errors: + - name: input_too_large + value: goccia.json.input_too_large + - name: null_rejected + value: goccia.json.null_rejected + - name: trailing_value + value: goccia.json.trailing_value + - name: syntax_error + value: goccia.json.syntax_error + - name: type_error + value: goccia.json.type_error + - name: unknown_field + value: goccia.json.unknown_field + + - name: json_encoder + errors: + - name: unsupported_type + value: goccia.json.unsupported_type + - name: unsupported_value + value: goccia.json.unsupported_value + - name: marshaler_error + value: goccia.json.marshaler_error + +groups: + - name: json_decoder + metrics: + - name: goccia.json.decoder.operation.duration + type: histogram + data_type: float + unit: s + error_set: json_decoder + + - name: goccia.json.decoder.input.size + type: histogram + unit: By + error_set: json_decoder +``` + +An inline form should also be possible: + +```yaml +errors: + - name: timeout + value: example.timeout + - name: cancelled + value: example.cancelled +``` + +Prefer `name` plus `value` over plain strings. `name` provides a stable Go +identifier while `value` remains the emitted OpenTelemetry value. + +## Proposed specification types + +Use Go names that make the OpenTelemetry concept explicit, even though the +YAML remains concise: + +```go +type ErrorType struct { + Name string `yaml:"name"` + Value string `yaml:"value"` + Description string `yaml:"description"` +} + +type ErrorSet struct { + Name string `yaml:"name"` + Errors []*ErrorType `yaml:"errors"` +} + +type Metric struct { + // Existing fields... + Errors []*ErrorType `yaml:"errors"` + ErrorSet string `yaml:"error_set"` +} + +type Spec struct { + // Existing fields... + ErrorSets []*ErrorSet `yaml:"error_sets"` +} +``` + +Decide and document whether inline `errors` and `error_set` may be combined. +The simplest first version should make them mutually exclusive. + +## Generated error types + +Named error sets should generate typed constants once, not once per metric. +For example, generate an `error_types.metrics.go` file: + +```go +type JsonDecoderErrorType string + +const ( + JsonDecoderErrorTypeInputTooLarge JsonDecoderErrorType = "goccia.json.input_too_large" + JsonDecoderErrorTypeNullRejected JsonDecoderErrorType = "goccia.json.null_rejected" + JsonDecoderErrorTypeTrailingValue JsonDecoderErrorType = "goccia.json.trailing_value" + JsonDecoderErrorTypeSyntaxError JsonDecoderErrorType = "goccia.json.syntax_error" + JsonDecoderErrorTypeTypeError JsonDecoderErrorType = "goccia.json.type_error" + JsonDecoderErrorTypeUnknownField JsonDecoderErrorType = "goccia.json.unknown_field" + JsonDecoderErrorTypeOther JsonDecoderErrorType = "_OTHER" +) +``` + +Generate `_OTHER` automatically for every set. Reject user entries named or +valued as the generated fallback unless the design explicitly changes to let +users declare it themselves. + +For inline errors, derive the type name from the metric title. Named sets are +preferred because inline metric-derived type names may be long. + +## Generated recording API + +A histogram with errors should get three APIs: + +```go +// Success: no error.type attribute. +func (m *JsonDecoder) RecordGocciaJsonDecoderOperationDuration( + ctx context.Context, + value float64, +) + +// Failure: records a validated error.type. +func (m *JsonDecoder) RecordGocciaJsonDecoderOperationDurationWithErrorType( + ctx context.Context, + value float64, + errorType JsonDecoderErrorType, +) + +// Existing escape hatch for callers that already own a complete attribute set. +func (m *JsonDecoder) RecordGocciaJsonDecoderOperationDurationWithAttributes( + ctx context.Context, + value float64, + attributes metric.MeasurementOption, +) +``` + +The ordinary `Record...` method must not accept an error string. This makes the +fast and semantically correct success path the default. + +If a metric also has ordinary attributes, keep those arguments on both the +success and error methods, with `errorType` as the final argument. + +## Why cache error attributes + +The naming is easy to misread: + +- OpenTelemetry's `metric.WithAttributes(...)` constructs an attribute set and + does non-trivial work on every call. +- Goccia's `RecordWithAttributes(...)` accepts an already-built + `metric.MeasurementOption`; it does not itself construct the attributes. + +Today, `telemetry.FloatHistogram.Record(ctx, value, dynamicAttrs...)` passes two +options to OpenTelemetry: the cached stage attributes and a freshly created +dynamic option. OpenTelemetry must allocate/build the dynamic set and merge it +with the stage set for every measurement. + +Instead, build one complete option per known error during `InitMetrics`: + +```go +m.jsonDecoderErrorTypeAttrs = map[JsonDecoderErrorType]metric.MeasurementOption{ + JsonDecoderErrorTypeSyntaxError: tel.NewMetricAttributes( + attribute.String( + "error.type", + string(JsonDecoderErrorTypeSyntaxError), + ), + ), + // Other known errors... +} +``` + +`tel.NewMetricAttributes` includes `stage.kind` and `stage.name`, so the cached +option is complete. The recording method can pass one option and avoid a +per-record set construction and merge: + +```go +func (m *JsonDecoder) RecordGocciaJsonDecoderOperationDurationWithErrorType( + ctx context.Context, + value float64, + errorType JsonDecoderErrorType, +) { + attrs, ok := m.jsonDecoderErrorTypeAttrs[errorType] + if !ok { + attrs = m.jsonDecoderErrorTypeAttrs[JsonDecoderErrorTypeOther] + } + + m.gocciaJsonDecoderOperationDuration.RecordWithAttributes(ctx, value, attrs) +} +``` + +Cache one map per unique error set within a generated metric group and reuse it +across all histograms in that group. A measurement option is attribute-specific, +not instrument-specific. + +The success path remains the fastest path: + +```go +m.gocciaJsonDecoderOperationDuration.Record(ctx, value) +``` + +This uses the histogram wrapper's existing cached base option. + +### Metrics that also have ordinary attributes + +Arbitrary ordinary attribute values cannot be fully cached. For the first +implementation, choose one of these policies and test it: + +1. Optimize only metrics with no ordinary attributes. For metrics with both + ordinary attributes and errors, build the complete combined set per call. +2. Temporarily reject `errors`/`error_set` together with + `attributes`/`attribute_set` and add combined support later. + +Option 1 is more general; option 2 is a smaller initial change. The JSON +metrics need only the optimized no-ordinary-attributes path. + +## Validation rules + +- Error-set names must be non-empty and unique. +- Error names must be non-empty and unique within their effective set. +- Error values must be non-empty and unique within their effective set. +- `error_set` must reference an existing set. +- Inline `errors` and `error_set` should initially be mutually exclusive. +- `_OTHER` is reserved if the generator adds it automatically. +- Initially support errors only for histogram metrics. Current counters are + observable counters and cannot attach attributes to individual increments. +- Generated Go identifiers must be unique after camel-case conversion. +- Unknown typed values passed at runtime must use the cached `_OTHER` option, + never create a new time series from the unknown string. + +## Generator changes + +- [ ] Extend `pkg/spec.go` with `ErrorType`, `ErrorSet`, metric error fields, + and top-level `error_sets`. +- [ ] Add an error-set registry and validation/resolution in + `pkg/validator.go`. +- [ ] Add templates for shared error types and constants. +- [ ] Extend `metricsFile`/generator template data with the effective error + types used by each group. +- [ ] Generate one cached attribute-option map per unique error set used by a + group. +- [ ] Update the histogram template with success and `WithErrorType` methods. +- [ ] Preserve the existing `WithAttributes` method. +- [ ] Add required `attribute` and `metric` imports when errors are configured. +- [ ] Include allowed error types in generated Markdown documentation. + +## JSON migration + +- [ ] Replace the `error_type` attribute set in + `processor/metrics/spec.yaml` with `json_encoder` and `json_decoder` + error sets. +- [ ] Regenerate processor metric files. +- [ ] Remove the handwritten JSON error-string constants from + `processor/json.go`; use the generated typed constants as the source of + truth. +- [ ] Change `getJSONErrorType` to return the generated encoder/decoder error + type. If one function cannot cleanly return two distinct set types, use a + common JSON error set or split it into encoder and decoder classifiers. +- [ ] On success, call the generated ordinary `Record...` method. +- [ ] On failure, call `Record...WithErrorType`. +- [ ] Apply the same conditional behavior to decoder input size. +- [ ] Remove the temporary `TODO! fix error type to not be included on success` + comments from the JSON workers. + +## Tests and benchmarks + +### Generator tests + +- [ ] Load and resolve valid inline errors and named error sets. +- [ ] Reject missing names/values, duplicates, unknown sets, reserved `_OTHER`, + and invalid metric types. +- [ ] Assert constants are emitted once when several metrics reuse a set. +- [ ] Assert success methods do not accept or record `error.type`. +- [ ] Assert error methods use the typed set and `_OTHER` fallback. +- [ ] Assert cached options include the base stage attributes. +- [ ] Keep the bucket-bound regression test passing. + +### Processor/telemetry tests + +- [ ] Successful JSON encode/decode measurements have no `error.type`. +- [ ] Failed measurements contain the expected classified value. +- [ ] Unknown failures use `_OTHER`. +- [ ] `stage.kind` and `stage.name` remain present on cached error measurements. + +### Benchmarks + +Compare at least these paths with `-benchmem` and a real SDK meter provider: + +1. success using cached base attributes; +2. failure using dynamically constructed attributes; +3. failure using the generated cached combined option. + +The expected order is success, cached failure, dynamic failure. Do not require +zero total allocations from the SDK; verify that the cached path removes the +extra attribute construction/merge cost. + +## Completion checklist + +- [ ] `go generate ./processor/metrics` is deterministic and idempotent. +- [ ] `go test ./...` passes from the repository root. +- [ ] `go test ./...` passes from `cmd/metrics-gen` (it is a separate module). +- [ ] `git diff --check` passes. +- [ ] Generated JSON success series omit `error.type` entirely. +- [ ] Generated error series use only declared values or `_OTHER`. + diff --git a/cmd/metrics-gen/pkg/generator.go b/cmd/metrics-gen/pkg/generator.go index e45ba63..cb51953 100644 --- a/cmd/metrics-gen/pkg/generator.go +++ b/cmd/metrics-gen/pkg/generator.go @@ -4,12 +4,14 @@ import ( "bytes" "fmt" "go/format" + "maps" "os" "path" + "path/filepath" + "slices" "text/template" "github.com/FerroO2000/goccia/cmd/metrics-gen/templates" - md "github.com/nao1215/markdown" ) var metricFileTmpl = template.Must( @@ -23,25 +25,51 @@ var metricFileTmpl = template.Must( ParseFS(templates.Templates, "*.tmpl"), ) +var errorTypesFileTmpl = template.Must( + template.New("error_types_file.go.tmpl"). + Funcs(template.FuncMap{ + "dict": dict, + "toUpperCamelCase": toUpperCamelCase, + "toLowerCamelCase": toLowerCamelCase, + "getDataType": getDataType, + }). + ParseFS(templates.Templates, "*.tmpl"), +) + var defaultImports = []string{"github.com/FerroO2000/goccia/internal/telemetry"} type metricsFile struct { - Name string - Package string - Imports []string - Metrics []*Metric + Name string + Package string + Imports []string + Metrics []*Metric + ErrorTypes []*ErrorType +} + +type errorTypesFile struct { + Package string + ErrorTypes []*ErrorType } // Generator struct defines a metrics generator. type Generator struct { basePath string + + markdown *markdownGenerator } // NewGenerator returns a new metrics generator instance // that writes files to the given base path. func NewGenerator(basePath string) *Generator { + mdBasePath := filepath.Join(basePath, "docs") + if err := os.MkdirAll(mdBasePath, os.ModePerm); err != nil { + panic(err) + } + return &Generator{ basePath: basePath, + + markdown: newMarkdownGenerator(mdBasePath), } } @@ -61,7 +89,7 @@ func (g *Generator) getImportPackages(metricType *Metric) []string { imports = append(imports, "go.opentelemetry.io/otel/metric") } - if len(metricType.Attributes) > 0 { + if len(metricType.Attributes) > 0 || metricType.ErrorTypeRef != nil { imports = append(imports, "go.opentelemetry.io/otel/attribute") } @@ -95,22 +123,41 @@ func (g *Generator) getMetricsFileName(name string) string { // Generate generates metrics files from the given spec. func (g *Generator) Generate(spec *Spec) error { for _, group := range spec.Groups { + errTypes := make(map[string]*ErrorType) + for _, metric := range group.Metrics { + if metric.ErrorTypeRef != nil { + errTypes[metric.ErrorTypeRef.Name] = metric.ErrorTypeRef + } + } + metricFile := &metricsFile{ - Name: group.Name, - Package: spec.Package, - Imports: g.getImports(group.Metrics), - Metrics: group.Metrics, + Name: group.Name, + Package: spec.Package, + Imports: g.getImports(group.Metrics), + Metrics: group.Metrics, + ErrorTypes: slices.Collect(maps.Values(errTypes)), } if err := g.generateMetricsFile(metricFile); err != nil { return err } + } + + if len(spec.ErrorTypes) > 0 { + errorFile := &errorTypesFile{ + Package: spec.Package, + ErrorTypes: spec.ErrorTypes, + } - if err := g.generateMarkdownFile(metricFile); err != nil { + if err := g.generateErrorTypesFile(errorFile); err != nil { return err } } + if err := g.markdown.generate(spec.Groups); err != nil { + return err + } + return nil } @@ -141,39 +188,29 @@ func (g *Generator) generateMetricsFile(mf *metricsFile) error { return nil } -func (g *Generator) getMarkdownFileName(name string) string { - fileName := toLowerSnakeCase(name) + ".doc.md" - return path.Join(g.basePath, fileName) -} +func (g *Generator) generateErrorTypesFile(ef *errorTypesFile) error { + var buf bytes.Buffer -func (g *Generator) generateMarkdownFile(mf *metricsFile) error { - file, err := os.Create(g.getMarkdownFileName(mf.Name)) - if err != nil { - return err + if err := errorTypesFileTmpl.ExecuteTemplate(&buf, "error_types_file.go.tmpl", ef); err != nil { + return fmt.Errorf("execute template: %w", err) } - defer file.Close() - rows := make([][]string, 0, len(mf.Metrics)) - for _, metric := range mf.Metrics { - typ := md.Code(string(metric.Type)) - dataType := md.Code(string(metric.DataType)) - desc := "-" - if metric.Description != "" { - desc = metric.Description - } + // Format as valid Go source + formatted, err := format.Source(buf.Bytes()) + if err != nil { + return fmt.Errorf("format source (raw output: %s): %w", buf.String(), err) + } - rows = append(rows, []string{ - metric.Name, - typ, - dataType, - desc, - }) + file, err := os.Create("error_types.metrics.go") + if err != nil { + return fmt.Errorf("create file: %w", err) } + defer file.Close() - mdFile := md.NewMarkdown(file).Table(md.TableSet{ - Header: []string{"Name", "Type", "Data Type", "Description"}, - Rows: rows, - }) + _, err = file.Write(formatted) + if err != nil { + return fmt.Errorf("write file: %w", err) + } - return mdFile.Build() + return nil } diff --git a/cmd/metrics-gen/pkg/generator_md.go b/cmd/metrics-gen/pkg/generator_md.go new file mode 100644 index 0000000..d088395 --- /dev/null +++ b/cmd/metrics-gen/pkg/generator_md.go @@ -0,0 +1,188 @@ +package pkg + +import ( + "fmt" + "os" + "path" + "strings" + + md "github.com/nao1215/markdown" +) + +type markdownGenerator struct { + basePath string +} + +func newMarkdownGenerator(basePath string) *markdownGenerator { + return &markdownGenerator{ + basePath: basePath, + } +} + +func (g *markdownGenerator) getFilename(name, typ string) string { + filename := fmt.Sprintf("%s.%s.doc.md", toLowerSnakeCase(name), typ) + return path.Join(g.basePath, filename) +} + +func (g *markdownGenerator) generate(groups []*Group) error { + for _, group := range groups { + metricsFilename := g.getFilename(group.Name, "metrics") + if err := g.generateMetricsTable(metricsFilename, group.Metrics); err != nil { + return err + } + + attributesFilename := g.getFilename(group.Name, "attributes") + if err := g.generateAttributesTable(attributesFilename, group.Metrics); err != nil { + return err + } + + errorTypesFilename := g.getFilename(group.Name, "error_types") + if err := g.generateErrorTypeTable(errorTypesFilename, group.Metrics); err != nil { + return err + } + } + + return nil +} + +func (g *markdownGenerator) getDescription(desc string) string { + if len(desc) == 0 { + return "-" + } + return desc +} + +func (g *markdownGenerator) getAttributeURL(attr *Attribute) string { + return fmt.Sprintf("#%s", toLowerSnakeCase(attr.Name)) +} + +func (g *markdownGenerator) joinItems(items ...string) string { + if len(items) == 0 { + return "-" + } + + strs := make([]string, 0, len(items)) + for _, item := range items { + strs = append(strs, item) + } + + return strings.Join(strs, " ") +} + +func (g *markdownGenerator) generateMetricsTable(filename string, metrics []*Metric) error { + if len(metrics) == 0 { + return nil + } + + file, err := os.Create(filename) + if err != nil { + return err + } + defer file.Close() + + rows := make([][]string, 0, len(metrics)) + for _, metric := range metrics { + name := metric.Name + typ := g.joinItems(md.Code(metric.Type), md.Code(metric.DataType)) + + attrsCollector := make([]string, 0, len(metric.Attributes)) + for _, attr := range metric.Attributes { + attrsCollector = append(attrsCollector, md.Link(attr.Name, g.getAttributeURL(attr))) + } + + if metric.ErrorTypeRef != nil { + attrsCollector = append(attrsCollector, md.Link("error.type", "#error_type")) + } + + attrs := g.joinItems(attrsCollector...) + + desc := g.getDescription(metric.Description) + + rows = append(rows, []string{ + name, typ, attrs, desc, + }) + } + + mdFile := md.NewMarkdown(file).Table(md.TableSet{ + Header: []string{"Name", "Type", "Attributes", "Description"}, + Rows: rows, + }) + + return mdFile.Build() +} + +func (g *markdownGenerator) generateAttributesTable(filename string, metrics []*Metric) error { + accumulator := make(map[string]*Attribute) + for _, metric := range metrics { + for _, attr := range metric.Attributes { + accumulator[attr.Name] = attr + } + } + + totAttributes := len(accumulator) + if totAttributes == 0 { + return nil + } + + file, err := os.Create(filename) + if err != nil { + return err + } + defer file.Close() + + rows := make([][]string, 0, totAttributes) + for _, attr := range accumulator { + name := fmt.Sprintf("`%s` {%s}", attr.Name, g.getAttributeURL(attr)) + typ := md.Code(attr.Type) + desc := g.getDescription(attr.Description) + + rows = append(rows, []string{name, typ, desc}) + } + + mdFile := md.NewMarkdown(file).Table(md.TableSet{ + Header: []string{"Name", "Type", "Description"}, + Rows: rows, + }) + + return mdFile.Build() +} + +func (g *markdownGenerator) generateErrorTypeTable(filename string, metrics []*Metric) error { + var errorType *ErrorType + for _, metric := range metrics { + if metric.ErrorTypeRef != nil { + errorType = metric.ErrorTypeRef + break + } + } + + if errorType == nil || len(errorType.Errors) == 0 { + return nil + } + + file, err := os.Create(filename) + if err != nil { + return err + } + defer file.Close() + + rows := make([][]string, 0, len(errorType.Errors)) + for idx, err := range errorType.Errors { + name := err.Name + if idx == 0 { + name = fmt.Sprintf("`%s` {#error_type}", err.Name) + } + + value := md.Code(err.Value) + desc := g.getDescription(err.Description) + + rows = append(rows, []string{name, value, desc}) + } + + mdFile := md.NewMarkdown(file).Table(md.TableSet{ + Header: []string{"Name", "Value", "Description"}, + Rows: rows, + }) + + return mdFile.Build() +} diff --git a/cmd/metrics-gen/pkg/spec.go b/cmd/metrics-gen/pkg/spec.go index e7b1fc0..040c86c 100644 --- a/cmd/metrics-gen/pkg/spec.go +++ b/cmd/metrics-gen/pkg/spec.go @@ -17,9 +17,10 @@ const ( // Attribute defines an attribute. type Attribute struct { - Name string `yaml:"name"` - Type AttributeType `yaml:"type"` - Arg string `yaml:"arg"` + Name string `yaml:"name"` + Type AttributeType `yaml:"type"` + Description string `yaml:"description"` + Arg string `yaml:"arg"` } func (a *Attribute) getName() string { @@ -52,15 +53,18 @@ const ( // Metric defines a metric. type Metric struct { - Name string `yaml:"name"` - Description string `yaml:"description"` - Type MetricType `yaml:"type"` - DataType DataType `yaml:"data_type"` - Unit string `yaml:"unit"` - BucketBounds []float64 `yaml:"bucket_bounds"` - CustomGetter bool `yaml:"custom_getter"` - Attributes []*Attribute `yaml:"attributes"` - AttributeSet string `yaml:"attribute_set"` + Name string `yaml:"name"` + Description string `yaml:"description"` + Type MetricType `yaml:"type"` + DataType DataType `yaml:"data_type"` + Unit string `yaml:"unit"` + CustomGetter bool `yaml:"custom_getter"` + Attributes []*Attribute `yaml:"attributes"` + AttributeSet string `yaml:"attribute_set"` + BucketBounds []float64 `yaml:"bucket_bounds"` + BucketBoundsSet string `yaml:"bucket_bounds_set"` + ErrorType string `yaml:"error_type"` + ErrorTypeRef *ErrorType } func (m *Metric) getName() string { @@ -84,9 +88,39 @@ type AttributeSet struct { Attributes []*Attribute `yaml:"attributes"` } +// BucketBoundsSet defines a set of bucket bounds to be referenced +// by other metrics. +type BucketBoundsSet struct { + Name string `yaml:"name"` + LowerBound float64 `yaml:"lower_bound"` + UpperBound float64 `yaml:"upper_bound"` + Bounds []float64 `yaml:"bounds"` +} + +// Error defines the value the error.type attribute +// of a metric can have. +type Error struct { + Name string `yaml:"name"` + Value string `yaml:"value"` + Description string `yaml:"description"` +} + +func (e *Error) getName() string { + return e.Name +} + +// ErrorType defines a set of errors to be referenced +// by other metrics. +type ErrorType struct { + Name string `yaml:"name"` + Errors []*Error `yaml:"errors"` +} + // Spec defines a metrics file spec. type Spec struct { - Package string `yaml:"package"` - AttributeSets []*AttributeSet `yaml:"attribute_sets"` - Groups []*Group `yaml:"groups"` + Package string `yaml:"package"` + AttributeSets []*AttributeSet `yaml:"attribute_sets"` + BucketBoundsSets []*BucketBoundsSet `yaml:"bucket_bounds_sets"` + ErrorTypes []*ErrorType `yaml:"error_types"` + Groups []*Group `yaml:"groups"` } diff --git a/cmd/metrics-gen/pkg/validator.go b/cmd/metrics-gen/pkg/validator.go index 3ced01f..c1ac338 100644 --- a/cmd/metrics-gen/pkg/validator.go +++ b/cmd/metrics-gen/pkg/validator.go @@ -1,6 +1,10 @@ package pkg -import "fmt" +import ( + "fmt" + "math" + "strconv" +) func isEmpty(field string) bool { return field == "" @@ -10,6 +14,35 @@ func requiredFieldErr(kind, fieldName string) error { return fmt.Errorf("%s: field '%s' is required", kind, fieldName) } +func calculateBucketBounds(lower, upper float64) []float64 { + mantissas := [...]float64{1, 2.5, 5} + + exponent := int(math.Floor(math.Log10(lower))) + var bounds []float64 + + for { + for _, mantissa := range mantissas { + // Construct the decimal value directly. Multiplying a mantissa by + // math.Pow can land one ULP away from the intended decimal value + // (for example, 2.4999999999999998e-06 instead of 2.5e-06). + literal := strconv.FormatFloat(mantissa, 'f', -1, 64) + "e" + strconv.Itoa(exponent) + bound, _ := strconv.ParseFloat(literal, 64) + + if bound < lower { + continue + } + + if bound > upper { + return bounds + } + + bounds = append(bounds, bound) + } + + exponent++ + } +} + type namedListItem interface { getName() string } @@ -41,14 +74,18 @@ func validateNamedList[T namedListItem](kind string, items []T, validateFn func( type specValidator struct { spec *Spec - attributeSets map[string]*AttributeSet + attributeSets map[string]*AttributeSet + bucketBoundsSets map[string]*BucketBoundsSet + errorTypes map[string]*ErrorType } func newSpecValidator(spec *Spec) *specValidator { return &specValidator{ spec: spec, - attributeSets: make(map[string]*AttributeSet, len(spec.AttributeSets)), + attributeSets: make(map[string]*AttributeSet, len(spec.AttributeSets)), + bucketBoundsSets: make(map[string]*BucketBoundsSet, len(spec.BucketBoundsSets)), + errorTypes: make(map[string]*ErrorType, len(spec.ErrorTypes)), } } @@ -63,6 +100,18 @@ func (v *specValidator) validate() error { } } + for _, bucketBoundsSet := range v.spec.BucketBoundsSets { + if err := v.validateBucketBoundsSet(bucketBoundsSet); err != nil { + return err + } + } + + for _, errType := range v.spec.ErrorTypes { + if err := v.validateErrorType(errType); err != nil { + return err + } + } + if err := validateNamedList("group", v.spec.Groups, v.validateGroup); err != nil { return err } @@ -104,6 +153,69 @@ func (v *specValidator) validateAttributeSet(attributeSet *AttributeSet) error { return nil } +func (v *specValidator) validateBucketBoundsSet(bucketBoundsSet *BucketBoundsSet) error { + if isEmpty(bucketBoundsSet.Name) { + return requiredFieldErr("bucket_bounds_set", "name") + } + + if _, ok := v.bucketBoundsSets[bucketBoundsSet.Name]; ok { + return fmt.Errorf("duplicated bucket bounds set name '%s'", bucketBoundsSet.Name) + } + + lower := bucketBoundsSet.LowerBound + upper := bucketBoundsSet.UpperBound + if lower != 0 && upper != 0 { + // Calculate the bounds + if lower < 0 { + return fmt.Errorf("invalid lower bound '%f'", lower) + } + + if upper < 0 { + return fmt.Errorf("invalid upper bound '%f'", upper) + } + + if upper < lower { + return fmt.Errorf("upper bound '%f' is less than lower bound '%f'", upper, lower) + } + + bucketBoundsSet.Bounds = calculateBucketBounds(lower, upper) + } + + v.bucketBoundsSets[bucketBoundsSet.Name] = bucketBoundsSet + + return nil +} + +func (v *specValidator) validateError(err *Error) error { + if isEmpty(err.Name) { + return requiredFieldErr("error", "name") + } + + if isEmpty(err.Value) { + return requiredFieldErr("error", "value") + } + + return nil +} + +func (v *specValidator) validateErrorType(errType *ErrorType) error { + if isEmpty(errType.Name) { + return requiredFieldErr("error_type", "name") + } + + if _, ok := v.errorTypes[errType.Name]; ok { + return fmt.Errorf("duplicated error type name '%s'", errType.Name) + } + + if err := validateNamedList("error", errType.Errors, v.validateError); err != nil { + return err + } + + v.errorTypes[errType.Name] = errType + + return nil +} + func (v *specValidator) validateMetric(metric *Metric) error { if isEmpty(metric.Name) { return requiredFieldErr("metric", "name") @@ -130,6 +242,24 @@ func (v *specValidator) validateMetric(metric *Metric) error { return err } + if len(metric.BucketBounds) == 0 && !isEmpty(metric.BucketBoundsSet) { + set, ok := v.bucketBoundsSets[metric.BucketBoundsSet] + if !ok { + return fmt.Errorf("unknown bucket bounds set '%s'", metric.BucketBoundsSet) + } + + metric.BucketBounds = set.Bounds + } + + if !isEmpty(metric.ErrorType) { + errType, ok := v.errorTypes[metric.ErrorType] + if !ok { + return fmt.Errorf("unknown error type '%s'", metric.ErrorType) + } + + metric.ErrorTypeRef = errType + } + return nil } diff --git a/cmd/metrics-gen/pkg/validator_test.go b/cmd/metrics-gen/pkg/validator_test.go new file mode 100644 index 0000000..d2a5284 --- /dev/null +++ b/cmd/metrics-gen/pkg/validator_test.go @@ -0,0 +1,38 @@ +package pkg + +import ( + "reflect" + "testing" +) + +func TestCalculateBucketBounds(t *testing.T) { + want := []float64{ + 0.000001, + 0.0000025, + 0.000005, + 0.00001, + 0.000025, + 0.00005, + 0.0001, + 0.00025, + 0.0005, + 0.001, + 0.0025, + 0.005, + 0.01, + 0.025, + 0.05, + 0.1, + 0.25, + 0.5, + 1, + 2.5, + 5, + 10, + } + + got := calculateBucketBounds(0.000001, 10) + if !reflect.DeepEqual(got, want) { + t.Fatalf("calculateBucketBounds() = %#v, want %#v", got, want) + } +} diff --git a/cmd/metrics-gen/templates/error_types_file.go.tmpl b/cmd/metrics-gen/templates/error_types_file.go.tmpl new file mode 100644 index 0000000..69fc8de --- /dev/null +++ b/cmd/metrics-gen/templates/error_types_file.go.tmpl @@ -0,0 +1,17 @@ +// Code generated by metrics-gen. DO NOT EDIT. + +package {{ .Package }} + +{{ range .ErrorTypes -}} +{{ $errType := print ( .Name | toUpperCamelCase ) "ErrorType" -}} +type {{ $errType}} string + +const ( +{{ range .Errors -}} +{{ $err := print $errType ( .Name | toUpperCamelCase) -}} + {{ $err }} {{ $errType }} = "{{ .Value }}" +{{ end -}} + + {{ $errType }}Other {{ $errType }} = "_OTHER" +) +{{ end -}} diff --git a/cmd/metrics-gen/templates/helpers.go.tmpl b/cmd/metrics-gen/templates/helpers.go.tmpl index 56fea09..1858754 100644 --- a/cmd/metrics-gen/templates/helpers.go.tmpl +++ b/cmd/metrics-gen/templates/helpers.go.tmpl @@ -68,4 +68,19 @@ func (m *{{ .StructType }}) Decrement{{ .TitleName }}() { {{ end -}} attribute.{{ $method }}("{{ .Name }}", {{ .Arg }}), {{ end -}} +{{ end -}} + +{{ define "errorTypeDefinition" -}} +{{ $type := print ( .Name | toUpperCamelCase ) "ErrorType" -}} +type {{ $type }} string + +const( +{{ range .ErrorTypes -}} +{{ $errType := print $type ( .Name | toUpperCamelCase ) -}} +{{ if .Description -}} + // {{ $errType }}: {{ .Description }} +{{ end -}} + {{ $errType }} {{ $type }} = "{{ .Value }}" +{{ end -}} +) {{ end -}} \ No newline at end of file diff --git a/cmd/metrics-gen/templates/histogram.go.tmpl b/cmd/metrics-gen/templates/histogram.go.tmpl index 11f6ac4..4c2875c 100644 --- a/cmd/metrics-gen/templates/histogram.go.tmpl +++ b/cmd/metrics-gen/templates/histogram.go.tmpl @@ -32,7 +32,7 @@ func (m *{{ .StructType }}) Record{{ .TitleName }}( } // Record{{ .TitleName }}WithAttributes records the given value -// ans attributes into the histogram metric. +// and attributes into the histogram metric. {{ template "methodDescComment" .Metric -}} func (m *{{ .StructType }}) Record{{ .TitleName }}WithAttributes( ctx context.Context, @@ -45,4 +45,22 @@ func (m *{{ .StructType }}) Record{{ .TitleName }}WithAttributes( attributes, ) } -{{- end }} \ No newline at end of file + +{{ if .Metric.ErrorTypeRef -}} +{{ $errTypeMap := print ( .Metric.ErrorTypeRef.Name | toLowerCamelCase ) "ErrorTypes" -}} +// Record{{ .TitleName }}WithErrorType records the given value +// and the error type into the histogram metric. +{{ template "methodDescComment" .Metric -}} +func (m *{{ .StructType }}) Record{{ .TitleName }}WithErrorType( + ctx context.Context, + value {{ .Metric.DataType | getDataType}}, + errorType {{ print ( .Metric.ErrorTypeRef.Name | toUpperCamelCase ) "ErrorType" }}, +) { + m.{{ .MetricVar }}.RecordWithAttributes( + ctx, + value, + m.{{ $errTypeMap }}[errorType], + ) +} +{{- end }} +{{- end }} diff --git a/cmd/metrics-gen/templates/metric_file.go.tmpl b/cmd/metrics-gen/templates/metric_file.go.tmpl index 8d0de31..9240824 100644 --- a/cmd/metrics-gen/templates/metric_file.go.tmpl +++ b/cmd/metrics-gen/templates/metric_file.go.tmpl @@ -25,6 +25,10 @@ type {{ $structType }} struct { {{- else if eq .Type "histogram" }} {{- template "histogramType" $entry -}} {{ end }}{{ end -}} + +{{ range .ErrorTypes }} + {{ .Name | toLowerCamelCase }}ErrorTypes map[{{ print ( .Name | toUpperCamelCase ) "ErrorType" }}]metric.MeasurementOption +{{ end }} } // New{{ $structType }} returns a new instance of the {{ $structType }} struct. @@ -58,6 +62,22 @@ func (m *{{ $structType }}) InitMetrics(tel *telemetry.Telemetry) error { return err } {{ end }} + +{{ range .ErrorTypes -}} +{{ $errType := print ( .Name | toUpperCamelCase ) "ErrorType" -}} +{{ $errTypeMap := print ( .Name | toLowerCamelCase ) "ErrorTypes" -}} + m.{{ $errTypeMap }} = map[{{ $errType }}]metric.MeasurementOption{ +{{ range .Errors -}} +{{ $err := print $errType ( .Name | toUpperCamelCase) -}} + {{ $err }}: metric.WithAttributes( + attribute.String("error.type", string({{ $err }})), + ), +{{ end -}} + {{ $errType}}Other: metric.WithAttributes( + attribute.String("error.type", "_OTHER"), + ), + } +{{ end -}} return nil } @@ -77,4 +97,4 @@ func (m *{{ $structType }}) InitMetrics(tel *telemetry.Telemetry) error { {{- else if eq .Type "histogram" }} {{ template "histogramMethods" $entry -}} {{- end }} -{{- end }} \ No newline at end of file +{{- end }} diff --git a/docs/pages/stages/egress/file.md b/docs/pages/stages/egress/file.md index b9cdb58..f165695 100644 --- a/docs/pages/stages/egress/file.md +++ b/docs/pages/stages/egress/file.md @@ -51,7 +51,7 @@ interval. For example, hourly rotation should include the hour in the path. ## Metrics ---8<-- "egress/metrics/file_stage.doc.md" +--8<-- "egress/metrics/docs/file_stage.metrics.doc.md" ## Internals diff --git a/docs/pages/stages/egress/questdb.md b/docs/pages/stages/egress/questdb.md index 5efc876..94c192c 100644 --- a/docs/pages/stages/egress/questdb.md +++ b/docs/pages/stages/egress/questdb.md @@ -49,7 +49,7 @@ This egress stage produces no downstream output message. ## Metrics ---8<-- "egress/metrics/quest_db_stage.doc.md" +--8<-- "egress/metrics/docs/quest_db_stage.metrics.doc.md" ## Internals diff --git a/docs/pages/stages/egress/tcp.md b/docs/pages/stages/egress/tcp.md index 049a5e6..4c7e72c 100644 --- a/docs/pages/stages/egress/tcp.md +++ b/docs/pages/stages/egress/tcp.md @@ -36,7 +36,7 @@ worker-pool config. ## Metrics ---8<-- "egress/metrics/tcp_stage.doc.md" +--8<-- "egress/metrics/docs/tcp_stage.metrics.doc.md" ## Internals diff --git a/docs/pages/stages/egress/udp.md b/docs/pages/stages/egress/udp.md index 50bdb75..d80d46d 100644 --- a/docs/pages/stages/egress/udp.md +++ b/docs/pages/stages/egress/udp.md @@ -36,7 +36,7 @@ This egress stage produces no downstream output message. ## Metrics ---8<-- "egress/metrics/udp_stage.doc.md" +--8<-- "egress/metrics/docs/udp_stage.metrics.doc.md" ## Internals diff --git a/docs/pages/stages/ingress/ebpf.md b/docs/pages/stages/ingress/ebpf.md index 4300a6c..3d7bbda 100644 --- a/docs/pages/stages/ingress/ebpf.md +++ b/docs/pages/stages/ingress/ebpf.md @@ -37,7 +37,7 @@ Additional interfaces: none. `EBPFMessage[T]` only implements the standard ## Metrics ---8<-- "ingress/metrics/ebpf_stage.doc.md" +--8<-- "ingress/metrics/docs/ebpf_stage.metrics.doc.md" ## Internals diff --git a/docs/pages/stages/ingress/file.md b/docs/pages/stages/ingress/file.md index e3260ce..b2bafa2 100644 --- a/docs/pages/stages/ingress/file.md +++ b/docs/pages/stages/ingress/file.md @@ -43,7 +43,7 @@ Additional interfaces: `message.Serializable`. `GetBytes()` returns `Chunk`. ## Metrics ---8<-- "ingress/metrics/file_stage.doc.md" +--8<-- "ingress/metrics/docs/file_stage.metrics.doc.md" ## Internals diff --git a/docs/pages/stages/ingress/http.md b/docs/pages/stages/ingress/http.md index 664045f..a1d5607 100644 --- a/docs/pages/stages/ingress/http.md +++ b/docs/pages/stages/ingress/http.md @@ -263,16 +263,11 @@ errors, including an address already in use, occur when the server starts in ## Metrics ---8<-- "ingress/metrics/http_stage.doc.md" +--8<-- "ingress/metrics/docs/http_stage.metrics.doc.md" -The HTTP semantic-convention histograms use these attributes: +### Attributes -| Attribute | Value | -| --- | --- | -| `http.request.method` | Request method. | -| `url.scheme` | `http` or `https`, inferred from the request's TLS state. | -| `http.response.status_code` | Status recorded by the response writer. | -| `network.protocol.version` | `1.0`, `1.1`, `2`, or another major/minor version reported by `net/http`. | +--8<-- "ingress/metrics/docs/http_stage.attributes.doc.md" `goccia.http.ingress.queue.wait.duration` has an `outcome` of `enqueued` or `rejected`. `goccia.http.ingress.response.wait.duration` uses the future state: diff --git a/docs/pages/stages/ingress/kafka.md b/docs/pages/stages/ingress/kafka.md index 2e3e8df..69e824f 100644 --- a/docs/pages/stages/ingress/kafka.md +++ b/docs/pages/stages/ingress/kafka.md @@ -57,7 +57,7 @@ Additional interfaces: `message.Serializable`. `GetBytes()` returns `Value`. ## Metrics ---8<-- "ingress/metrics/kafka_stage.doc.md" +--8<-- "ingress/metrics/docs/kafka_stage.metrics.doc.md" ## Internals diff --git a/docs/pages/stages/ingress/tcp.md b/docs/pages/stages/ingress/tcp.md index bc6a156..3dffc7a 100644 --- a/docs/pages/stages/ingress/tcp.md +++ b/docs/pages/stages/ingress/tcp.md @@ -55,7 +55,7 @@ For length-prefixed mode, set `HeaderLen`, `MessageLengthFieldLen`, ## Metrics ---8<-- "ingress/metrics/tcp_stage.doc.md" +--8<-- "ingress/metrics/docs/tcp_stage.metrics.doc.md" ## Internals diff --git a/docs/pages/stages/ingress/ticker.md b/docs/pages/stages/ingress/ticker.md index 0782563..fcfd325 100644 --- a/docs/pages/stages/ingress/ticker.md +++ b/docs/pages/stages/ingress/ticker.md @@ -32,7 +32,7 @@ Additional interfaces: none. `TickerMessage` only implements the standard ## Metrics ---8<-- "ingress/metrics/ticker_stage.doc.md" +--8<-- "ingress/metrics/docs/ticker_stage.metrics.doc.md" ## Internals diff --git a/docs/pages/stages/ingress/udp.md b/docs/pages/stages/ingress/udp.md index c0f06cd..1560fe2 100644 --- a/docs/pages/stages/ingress/udp.md +++ b/docs/pages/stages/ingress/udp.md @@ -37,7 +37,7 @@ Additional interfaces: `message.Serializable`. `GetBytes()` returns ## Metrics ---8<-- "ingress/metrics/udp_stage.doc.md" +--8<-- "ingress/metrics/docs/udp_stage.metrics.doc.md" ## Internals diff --git a/docs/pages/stages/processor/aggregate.md b/docs/pages/stages/processor/aggregate.md index 1636303..88d0f02 100644 --- a/docs/pages/stages/processor/aggregate.md +++ b/docs/pages/stages/processor/aggregate.md @@ -81,7 +81,7 @@ The stage is intentionally single-threaded and does not accept a running mode. ## Metrics ---8<-- "processor/metrics/aggregate_stage.doc.md" +--8<-- "processor/metrics/docs/aggregate_stage.metrics.doc.md" The common processor processed-message counter is incremented by the number of input messages placed into emitted batches. diff --git a/docs/pages/stages/processor/can.md b/docs/pages/stages/processor/can.md index 63ba514..aca4bac 100644 --- a/docs/pages/stages/processor/can.md +++ b/docs/pages/stages/processor/can.md @@ -48,7 +48,7 @@ Additional interfaces: none. `CANMessage` only implements the standard ## Metrics ---8<-- "processor/metrics/can_stage.doc.md" +--8<-- "processor/metrics/docs/can_stage.metrics.doc.md" ## Internals diff --git a/docs/pages/stages/processor/filter.md b/docs/pages/stages/processor/filter.md index a3764b4..78afbc2 100644 --- a/docs/pages/stages/processor/filter.md +++ b/docs/pages/stages/processor/filter.md @@ -45,7 +45,7 @@ argument, not through the config object. ## Metrics ---8<-- "processor/metrics/filter_stage.doc.md" +--8<-- "processor/metrics/docs/filter_stage.metrics.doc.md" ## Internals diff --git a/docs/pages/stages/processor/index.md b/docs/pages/stages/processor/index.md index 2057f87..e61ce30 100644 --- a/docs/pages/stages/processor/index.md +++ b/docs/pages/stages/processor/index.md @@ -16,6 +16,8 @@ Processor stages read from an input connector and write to an output connector. | Generic | [Router](router.md) | `processor.NewRouterStage` | Single runner | Route each message to one output | | CSV | [Decoder](csv-decoder.md) | `processor.NewCSVDecoderStage` | [Pool-capable](../../concepts/stages.md#pooled-execution-mode){ .stage-badge .stage-badge--pool } | Bytes to typed CSV rows | | CSV | [Encoder](csv-encoder.md) | `processor.NewCSVEncoderStage` | [Pool-capable](../../concepts/stages.md#pooled-execution-mode){ .stage-badge .stage-badge--pool } | Typed CSV rows to bytes | +| JSON | [Decoder](json-decoder.md) | `processor.NewJSONDecoderStage` | [Pool-capable](../../concepts/stages.md#pooled-execution-mode){ .stage-badge .stage-badge--pool } | JSON bytes to typed Go values | +| JSON | [Encoder](json-encoder.md) | `processor.NewJSONEncoderStage` | [Pool-capable](../../concepts/stages.md#pooled-execution-mode){ .stage-badge .stage-badge--pool } | Typed Go values to JSON bytes | | CAN | [CAN](can.md) | `processor.NewCANStage` | [Pool-capable](../../concepts/stages.md#pooled-execution-mode){ .stage-badge .stage-badge--pool } | Raw CAN frames to decoded signals | | Cannelloni | [Decoder](cannelloni-decoder.md) | `processor.NewCannelloniDecoderStage` | [Pool-capable](../../concepts/stages.md#pooled-execution-mode){ .stage-badge .stage-badge--pool } | Cannelloni bytes to CAN frames | | Cannelloni | [Encoder](cannelloni-encoder.md) | `processor.NewCannelloniEncoderStage` | [Pool-capable](../../concepts/stages.md#pooled-execution-mode){ .stage-badge .stage-badge--pool } | CAN frames to Cannelloni bytes | diff --git a/docs/pages/stages/processor/json-decoder.md b/docs/pages/stages/processor/json-decoder.md new file mode 100644 index 0000000..d7fde7d --- /dev/null +++ b/docs/pages/stages/processor/json-decoder.md @@ -0,0 +1,83 @@ +--- +icon: lucide/braces +--- + +# JSON Decoder Processor + +`JSONDecoderStage` converts serializable JSON bytes into typed Go values held +by `JSONMessage`. + +[Pool-capable](../../concepts/stages.md#pooled-execution-mode){ .stage-badge .stage-badge--pool } + +``` go +type Event struct { + Name string `json:"name"` + Count int `json:"count"` +} + +cfg := processor.NewJSONDecoderConfig(goccia.StageRunningModePool) +cfg.DisallowUnknownFields = true +stage := processor.NewJSONDecoderStage[InputMessage, *Event](in, out, cfg) +``` + +## Messages + +### Input Message + +Accepted body requirement: `message.Serializable`. + +The decoder reads the byte slice returned by `GetBytes()`. Leading and trailing +JSON whitespace is accepted and counts toward `MaxInputBytes`. + +### Output Message + +Produced body type: `*processor.JSONMessage[Out]`. + +| Field | Description | +| --- | --- | +| `Data` | Value decoded by Go's `encoding/json` package. | + +Additional interfaces: none. `JSONMessage` only implements the standard +`message.Body` contract. + +`Out` can be any type supported by `encoding/json`. A pointer type is useful +when the application needs to distinguish a decoded value from `null`: unless +`RejectNull` is enabled, a top-level `null` produces a nil pointer. + +## Configuration + +| Field | Default | Description | +| --- | --- | --- | +| `Stage.RunningMode` | constructor arg | `StageRunningModeSingle` or `StageRunningModePool`. | +| `Stage.Pool` | default pool when pool mode is selected | Worker counts, queue sizes, and auto-scaling. | +| `DisallowUnknownFields` | `false` | Reject unknown object keys when decoding into a struct. Maps and interface values are unaffected. | +| `UseNumber` | `false` | Decode numbers stored in interface values as `json.Number` instead of `float64`. Typed numeric fields are unaffected. | +| `MaxInputBytes` | `0` | Maximum input size, including surrounding whitespace. Zero disables the limit; negative values are reset to zero during validation. | +| `RejectNull` | `false` | Reject a top-level JSON `null`. Nested null values remain allowed. | + +Decoder-specific settings are captured during stage initialization. Update the +configuration before calling `Pipeline.Init`. + +The decoder accepts exactly one top-level JSON value. Empty, malformed, +truncated, type-incompatible, oversized, and disallowed inputs return an error +to the processor runner. The runner records the processing failure and applies +its configured error behavior. + +## Metrics + +--8<-- "processor/metrics/docs/json_decoder.metrics.doc.md" + +### Error Types + +--8<-- "processor/metrics/docs/json_decoder.error_types.doc.md" + +Successful measurements omit `error.type`. + +## Internals + +The stage uses Goccia's generic worker-backed processor runner and Go's +`encoding/json` package. The default path uses `json.Unmarshal`; enabling +`DisallowUnknownFields` or `UseNumber` switches to `json.Decoder`. Each output +message preserves the processing span for downstream stages. + +Continue to the [JSON encoder processor](json-encoder.md). diff --git a/docs/pages/stages/processor/json-encoder.md b/docs/pages/stages/processor/json-encoder.md new file mode 100644 index 0000000..84ea6fb --- /dev/null +++ b/docs/pages/stages/processor/json-encoder.md @@ -0,0 +1,84 @@ +--- +icon: lucide/braces +--- + +# JSON Encoder Processor + +`JSONEncoderStage` converts typed values held by `JSONMessage` into serialized +JSON bytes. + +[Pool-capable](../../concepts/stages.md#pooled-execution-mode){ .stage-badge .stage-badge--pool } + +``` go +type Event struct { + Name string `json:"name"` + Count int `json:"count"` +} + +cfg := processor.NewJSONEncoderConfig(goccia.StageRunningModePool) +stage := processor.NewJSONEncoderStage(in, out, cfg) +``` + +Create an input body with `processor.NewJSONMessage`: + +``` go +body := processor.NewJSONMessage(Event{Name: "ready", Count: 1}) +``` + +## Messages + +### Input Message + +Accepted body type: `*processor.JSONMessage[T]`. + +| Field | Description | +| --- | --- | +| `Data` | Go value passed to `encoding/json`. | + +Additional input interfaces: none beyond the standard `message.Body` contract. + +### Output Message + +Produced body type: `*processor.JSONEncodedMessage`. + +| Field | Description | +| --- | --- | +| `Data` | Encoded JSON bytes without a trailing newline. | + +Additional interfaces: `message.Serializable`. `GetBytes()` returns `Data` +without copying it; callers must not mutate the slice while the message is in +use. + +## Configuration + +| Field | Default | Description | +| --- | --- | --- | +| `Stage.RunningMode` | constructor arg | `StageRunningModeSingle` or `StageRunningModePool`. | +| `Stage.Pool` | default pool when pool mode is selected | Worker counts, queue sizes, and auto-scaling. | +| `Indent` | `""` | Pretty-print indentation inserted for each nesting level. An empty value emits compact JSON. | +| `IndentPrefix` | `""` | Prefix written at the beginning of indented lines. Ignored when `Indent` is empty. | +| `EscapeHTML` | `true` | Escape `<`, `>`, and `&` inside strings as JSON Unicode sequences. | + +Use only JSON whitespace in `Indent` and `IndentPrefix`; other characters can +produce output that is not valid JSON. Encoder-specific settings are captured +during stage initialization. + +## Metrics + +--8<-- "processor/metrics/docs/json_encoder.metrics.doc.md" + +### Error Types + +--8<-- "processor/metrics/docs/json_encoder.error_types.doc.md" + +Successful measurements omit `error.type`; output size is recorded only after +successful encoding. + +## Internals + +The stage uses Goccia's generic worker-backed processor runner and Go's +`encoding/json` package. Compact output with HTML escaping uses `json.Marshal`; +indented output uses `json.MarshalIndent`. Disabling HTML escaping uses +`json.Encoder`, and the encoder removes its terminating newline before emitting +the message. Each output message preserves the processing span for downstream +stages. diff --git a/docs/pages/stages/processor/rob.md b/docs/pages/stages/processor/rob.md index 60eefb4..46e7e8c 100644 --- a/docs/pages/stages/processor/rob.md +++ b/docs/pages/stages/processor/rob.md @@ -47,7 +47,7 @@ The stage is intentionally single-threaded and does not accept a running mode. ## Metrics ---8<-- "processor/metrics/rob_stage.doc.md" +--8<-- "processor/metrics/docs/rob_stage.metrics.doc.md" ## Internals diff --git a/docs/pages/stages/processor/router.md b/docs/pages/stages/processor/router.md index 0b6c6c6..9495207 100644 --- a/docs/pages/stages/processor/router.md +++ b/docs/pages/stages/processor/router.md @@ -60,7 +60,7 @@ the processor error and dropped-message metrics are incremented. ## Metrics ---8<-- "processor/metrics/router_stage.doc.md" +--8<-- "processor/metrics/docs/router_stage.metrics.doc.md" `RouterStage` also records `routed_messages_per_route`, a per-route message count with `route_id` and `route_name` attributes. diff --git a/docs/pages/stages/processor/tee.md b/docs/pages/stages/processor/tee.md index deeb961..a762589 100644 --- a/docs/pages/stages/processor/tee.md +++ b/docs/pages/stages/processor/tee.md @@ -35,7 +35,7 @@ the envelopes are destroyed. ## Metrics ---8<-- "processor/metrics/tee_stage.doc.md" +--8<-- "processor/metrics/docs/tee_stage.metrics.doc.md" ## Internals diff --git a/docs/zensical.toml b/docs/zensical.toml index 8cb684d..3ad108d 100644 --- a/docs/zensical.toml +++ b/docs/zensical.toml @@ -42,6 +42,10 @@ nav = [ { "Decoder" = "stages/processor/csv-decoder.md" }, { "Encoder" = "stages/processor/csv-encoder.md" }, ] }, + { "JSON" = [ + { "Decoder" = "stages/processor/json-decoder.md" }, + { "Encoder" = "stages/processor/json-encoder.md" }, + ] }, { "CAN" = "stages/processor/can.md" }, { "Cannelloni" = [ { "Decoder" = "stages/processor/cannelloni-decoder.md" }, diff --git a/egress/metrics/docs/file_stage.metrics.doc.md b/egress/metrics/docs/file_stage.metrics.doc.md new file mode 100644 index 0000000..4376a5d --- /dev/null +++ b/egress/metrics/docs/file_stage.metrics.doc.md @@ -0,0 +1,5 @@ +| Name | Type | Attributes | Description | +|---------|---------|---------|---------| +| written_bytes | `counter` `integer` | - | - | +| write_errors | `counter` `integer` | - | - | +| flush_errors | `counter` `integer` | - | - | diff --git a/egress/metrics/docs/quest_db_stage.metrics.doc.md b/egress/metrics/docs/quest_db_stage.metrics.doc.md new file mode 100644 index 0000000..83a75f0 --- /dev/null +++ b/egress/metrics/docs/quest_db_stage.metrics.doc.md @@ -0,0 +1,3 @@ +| Name | Type | Attributes | Description | +|---------|---------|---------|---------| +| inserted_rows | `counter` `integer` | - | - | diff --git a/egress/metrics/docs/tcp_stage.metrics.doc.md b/egress/metrics/docs/tcp_stage.metrics.doc.md new file mode 100644 index 0000000..2f270d7 --- /dev/null +++ b/egress/metrics/docs/tcp_stage.metrics.doc.md @@ -0,0 +1,3 @@ +| Name | Type | Attributes | Description | +|---------|---------|---------|---------| +| delivered_bytes | `counter` `integer` | - | - | diff --git a/egress/metrics/docs/udp_stage.metrics.doc.md b/egress/metrics/docs/udp_stage.metrics.doc.md new file mode 100644 index 0000000..2f270d7 --- /dev/null +++ b/egress/metrics/docs/udp_stage.metrics.doc.md @@ -0,0 +1,3 @@ +| Name | Type | Attributes | Description | +|---------|---------|---------|---------| +| delivered_bytes | `counter` `integer` | - | - | diff --git a/egress/metrics/file_stage.doc.md b/egress/metrics/file_stage.doc.md deleted file mode 100644 index d917f6a..0000000 --- a/egress/metrics/file_stage.doc.md +++ /dev/null @@ -1,5 +0,0 @@ -| Name | Type | Data Type | Description | -|---------|---------|---------|---------| -| written_bytes | `counter` | `integer` | - | -| write_errors | `counter` | `integer` | - | -| flush_errors | `counter` | `integer` | - | diff --git a/egress/metrics/quest_db_stage.doc.md b/egress/metrics/quest_db_stage.doc.md deleted file mode 100644 index 5cb5e41..0000000 --- a/egress/metrics/quest_db_stage.doc.md +++ /dev/null @@ -1,3 +0,0 @@ -| Name | Type | Data Type | Description | -|---------|---------|---------|---------| -| inserted_rows | `counter` | `integer` | - | diff --git a/egress/metrics/tcp_stage.doc.md b/egress/metrics/tcp_stage.doc.md deleted file mode 100644 index 48337ac..0000000 --- a/egress/metrics/tcp_stage.doc.md +++ /dev/null @@ -1,3 +0,0 @@ -| Name | Type | Data Type | Description | -|---------|---------|---------|---------| -| delivered_bytes | `counter` | `integer` | - | diff --git a/egress/metrics/udp_stage.doc.md b/egress/metrics/udp_stage.doc.md deleted file mode 100644 index 48337ac..0000000 --- a/egress/metrics/udp_stage.doc.md +++ /dev/null @@ -1,3 +0,0 @@ -| Name | Type | Data Type | Description | -|---------|---------|---------|---------| -| delivered_bytes | `counter` | `integer` | - | diff --git a/examples/http/handler.go b/examples/http/handler.go deleted file mode 100644 index 4cd492a..0000000 --- a/examples/http/handler.go +++ /dev/null @@ -1,30 +0,0 @@ -package main - -import ( - "context" - "net/http" - - "github.com/FerroO2000/goccia/egress" - "github.com/FerroO2000/goccia/ingress" - "github.com/FerroO2000/goccia/processor" -) - -type echoHandler struct { - processor.GenericHandlerBase -} - -func newEchoHandler() *echoHandler { - return &echoHandler{} -} - -func (h *echoHandler) Handle(_ context.Context, req *ingress.HTTPMessage) (*egress.HTTPMessage, error) { - return &egress.HTTPMessage{ - StatusCode: http.StatusOK, - Header: http.Header{ - "Content-Type": []string{"application/octet-stream"}, - "X-Echo-Method": []string{req.Method}, - "X-Echo-Path": []string{req.Path}, - }, - Body: req.Body, - }, nil -} diff --git a/examples/http/handlers.go b/examples/http/handlers.go new file mode 100644 index 0000000..379736c --- /dev/null +++ b/examples/http/handlers.go @@ -0,0 +1,45 @@ +package main + +import ( + "context" + "net/http" + "sync/atomic" + + "github.com/FerroO2000/goccia/egress" + "github.com/FerroO2000/goccia/ingress" + "github.com/FerroO2000/goccia/processor" +) + +type echoHandler struct { + processor.GenericHandlerBase + + count atomic.Int64 +} + +func newEchoHandler() *echoHandler { + return &echoHandler{} +} + +func (h *echoHandler) Handle(_ context.Context, _ *ingress.HTTPMessage) (*jsonMessage, error) { + return processor.NewJSONMessage(responseBody{ + EchoCount: int(h.count.Add(1)), + }), nil +} + +type httpResponseHandler struct { + processor.GenericHandlerBase +} + +func newHTTPResponseHandler() *httpResponseHandler { + return &httpResponseHandler{} +} + +func (h *httpResponseHandler) Handle(_ context.Context, jsonEnc *processor.JSONEncodedMessage) (*egress.HTTPMessage, error) { + return &egress.HTTPMessage{ + StatusCode: http.StatusOK, + Header: http.Header{ + "Content-Type": []string{"application/json"}, + }, + Body: jsonEnc.Data, + }, nil +} diff --git a/examples/http/main.go b/examples/http/main.go index b10c915..29f8c44 100644 --- a/examples/http/main.go +++ b/examples/http/main.go @@ -18,6 +18,12 @@ import ( const connectorSize = 512 +type responseBody struct { + EchoCount int `json:"echo_count"` +} + +type jsonMessage = processor.JSONMessage[responseBody] + func main() { ctx, cancelCtx := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer cancelCtx() @@ -26,7 +32,9 @@ func main() { defer telemetry.Close() httpIngressToEcho := connector.NewRingBuffer[*ingress.HTTPMessage](connectorSize) - echoToHTTPEgress := connector.NewRingBuffer[*egress.HTTPMessage](connectorSize) + echoToJSON := connector.NewRingBuffer[*jsonMessage](connectorSize) + jsonToHTTPResponse := connector.NewRingBuffer[*processor.JSONEncodedMessage](connectorSize) + httpResponseToEgress := connector.NewRingBuffer[*egress.HTTPMessage](connectorSize) httpLink := link.NewHTTP() @@ -34,15 +42,25 @@ func main() { httpIngressStage := ingress.NewHTTPStage(httpLink, httpIngressToEcho, httpIngressCfg) echoCfg := processor.NewGenericConfig(goccia.StageRunningModeSingle) - echoCfg.Name = "http_echo" - echoStage := processor.NewGenericStage(newEchoHandler(), httpIngressToEcho, echoToHTTPEgress, echoCfg) + echoCfg.Name = "echo" + echoStage := processor.NewGenericStage(newEchoHandler(), httpIngressToEcho, echoToJSON, echoCfg) + + jsonEncCfg := processor.NewJSONEncoderConfig(goccia.StageRunningModePool) + jsonEncStage := processor.NewJSONEncoderStage(echoToJSON, jsonToHTTPResponse, jsonEncCfg) + + httpResponseCfg := processor.NewGenericConfig(goccia.StageRunningModeSingle) + httpResponseCfg.Name = "http_response" + httpResponseStage := processor.NewGenericStage(newHTTPResponseHandler(), jsonToHTTPResponse, httpResponseToEgress, httpResponseCfg) httpEgressCfg := egress.NewHTTPConfig() - httpEgressStage := egress.NewHTTPStage(httpLink, echoToHTTPEgress, httpEgressCfg) + httpEgressStage := egress.NewHTTPStage(httpLink, httpResponseToEgress, httpEgressCfg) pipeline := goccia.NewPipeline() + pipeline.AddStage(httpIngressStage) pipeline.AddStage(echoStage) + pipeline.AddStage(jsonEncStage) + pipeline.AddStage(httpResponseStage) pipeline.AddStage(httpEgressStage) if err := pipeline.Init(ctx); err != nil { diff --git a/ingress/metrics/docs/ebpf_stage.metrics.doc.md b/ingress/metrics/docs/ebpf_stage.metrics.doc.md new file mode 100644 index 0000000..f92d04f --- /dev/null +++ b/ingress/metrics/docs/ebpf_stage.metrics.doc.md @@ -0,0 +1,5 @@ +| Name | Type | Attributes | Description | +|---------|---------|---------|---------| +| received_records | `counter` `integer` | - | - | +| parsing_errors | `counter` `integer` | - | - | +| received_bytes | `counter` `integer` | - | - | diff --git a/ingress/metrics/docs/file_stage.metrics.doc.md b/ingress/metrics/docs/file_stage.metrics.doc.md new file mode 100644 index 0000000..9ab41a0 --- /dev/null +++ b/ingress/metrics/docs/file_stage.metrics.doc.md @@ -0,0 +1,5 @@ +| Name | Type | Attributes | Description | +|---------|---------|---------|---------| +| readers | `upDownCounter` `integer` | - | - | +| active_readers | `upDownCounter` `integer` | - | - | +| read_bytes | `counter` `integer` | - | - | diff --git a/ingress/metrics/docs/http_stage.attributes.doc.md b/ingress/metrics/docs/http_stage.attributes.doc.md new file mode 100644 index 0000000..a2129bd --- /dev/null +++ b/ingress/metrics/docs/http_stage.attributes.doc.md @@ -0,0 +1,7 @@ +| Name | Type | Description | +|---------|---------|---------| +| `outcome` {#outcome} | `string` | - | +| `http.request.method` {#http.request.method} | `string` | Request method. | +| `url.scheme` {#url.scheme} | `string` | `http` or `https`, inferred from the request's TLS state. | +| `http.response.status_code` {#http.response.status_code} | `int` | Status recorded by the response writer. | +| `network.protocol.version` {#network.protocol.version} | `string` | `1.0`, `1.1`, `2`, or another major/minor version reported by `net/http`. | diff --git a/ingress/metrics/docs/http_stage.metrics.doc.md b/ingress/metrics/docs/http_stage.metrics.doc.md new file mode 100644 index 0000000..5c24e4f --- /dev/null +++ b/ingress/metrics/docs/http_stage.metrics.doc.md @@ -0,0 +1,10 @@ +| Name | Type | Attributes | Description | +|---------|---------|---------|---------| +| http.server.request.duration | `histogram` `float` | [http.request.method](#http.request.method) [url.scheme](#url.scheme) [http.response.status_code](#http.response.status_code) [network.protocol.version](#network.protocol.version) | Duration of HTTP server requests from handler entry until the response is written. | +| http.server.active_requests | `upDownCounter` `integer` | - | Number of HTTP server requests currently being handled. | +| http.server.request.body.size | `histogram` `integer` | [http.request.method](#http.request.method) [url.scheme](#url.scheme) [http.response.status_code](#http.response.status_code) [network.protocol.version](#network.protocol.version) | Size of HTTP request bodies observed by the server, including partially read bodies. | +| http.server.response.body.size | `histogram` `integer` | [http.request.method](#http.request.method) [url.scheme](#url.scheme) [http.response.status_code](#http.response.status_code) [network.protocol.version](#network.protocol.version) | Size of HTTP response bodies successfully written by the server. | +| goccia.http.ingress.queue.len | `gauge` `integer` | - | Number of HTTP request messages currently waiting in the ingress output queue. | +| goccia.http.ingress.queue.wait.duration | `histogram` `float` | [outcome](#outcome) | Time spent waiting to enqueue an HTTP request message for downstream processing. | +| goccia.http.ingress.pending_responses | `upDownCounter` `integer` | - | Number of HTTP requests currently awaiting a downstream response. | +| goccia.http.ingress.response.wait.duration | `histogram` `float` | [outcome](#outcome) | Time spent waiting for a downstream response after a successful queue handoff. | diff --git a/ingress/metrics/docs/kafka_stage.metrics.doc.md b/ingress/metrics/docs/kafka_stage.metrics.doc.md new file mode 100644 index 0000000..6e0a529 --- /dev/null +++ b/ingress/metrics/docs/kafka_stage.metrics.doc.md @@ -0,0 +1,4 @@ +| Name | Type | Attributes | Description | +|---------|---------|---------|---------| +| received_messages | `counter` `integer` | - | - | +| received_bytes | `counter` `integer` | - | - | diff --git a/ingress/metrics/docs/tcp_stage.metrics.doc.md b/ingress/metrics/docs/tcp_stage.metrics.doc.md new file mode 100644 index 0000000..a6a264e --- /dev/null +++ b/ingress/metrics/docs/tcp_stage.metrics.doc.md @@ -0,0 +1,5 @@ +| Name | Type | Attributes | Description | +|---------|---------|---------|---------| +| open_connections | `upDownCounter` `integer` | - | - | +| received_messages | `counter` `integer` | - | - | +| received_bytes | `counter` `integer` | - | - | diff --git a/ingress/metrics/docs/ticker_stage.metrics.doc.md b/ingress/metrics/docs/ticker_stage.metrics.doc.md new file mode 100644 index 0000000..b462480 --- /dev/null +++ b/ingress/metrics/docs/ticker_stage.metrics.doc.md @@ -0,0 +1,3 @@ +| Name | Type | Attributes | Description | +|---------|---------|---------|---------| +| triggered_messages | `counter` `integer` | - | - | diff --git a/ingress/metrics/docs/udp_stage.metrics.doc.md b/ingress/metrics/docs/udp_stage.metrics.doc.md new file mode 100644 index 0000000..6e0a529 --- /dev/null +++ b/ingress/metrics/docs/udp_stage.metrics.doc.md @@ -0,0 +1,4 @@ +| Name | Type | Attributes | Description | +|---------|---------|---------|---------| +| received_messages | `counter` `integer` | - | - | +| received_bytes | `counter` `integer` | - | - | diff --git a/ingress/metrics/ebpf_stage.doc.md b/ingress/metrics/ebpf_stage.doc.md index f30e6e5..ab37731 100644 --- a/ingress/metrics/ebpf_stage.doc.md +++ b/ingress/metrics/ebpf_stage.doc.md @@ -1,5 +1,8 @@ -| Name | Type | Data Type | Description | -|---------|---------|---------|---------| -| received_records | `counter` | `integer` | - | -| parsing_errors | `counter` | `integer` | - | -| received_bytes | `counter` | `integer` | - | +| Name | Type | Data Type | Attributes | Description | +|---------|---------|---------|---------|---------| +| received_records | `counter` | `integer` | - | - | +| parsing_errors | `counter` | `integer` | - | - | +| received_bytes | `counter` | `integer` | - | - | + +| Name | Type | +|---------|---------| diff --git a/ingress/metrics/ebpf_stage.metrics.doc.md b/ingress/metrics/ebpf_stage.metrics.doc.md new file mode 100644 index 0000000..f92d04f --- /dev/null +++ b/ingress/metrics/ebpf_stage.metrics.doc.md @@ -0,0 +1,5 @@ +| Name | Type | Attributes | Description | +|---------|---------|---------|---------| +| received_records | `counter` `integer` | - | - | +| parsing_errors | `counter` `integer` | - | - | +| received_bytes | `counter` `integer` | - | - | diff --git a/ingress/metrics/file_stage.doc.md b/ingress/metrics/file_stage.doc.md index 6ddbe1d..7b4332c 100644 --- a/ingress/metrics/file_stage.doc.md +++ b/ingress/metrics/file_stage.doc.md @@ -1,5 +1,8 @@ -| Name | Type | Data Type | Description | -|---------|---------|---------|---------| -| readers | `upDownCounter` | `integer` | - | -| active_readers | `upDownCounter` | `integer` | - | -| read_bytes | `counter` | `integer` | - | +| Name | Type | Data Type | Attributes | Description | +|---------|---------|---------|---------|---------| +| readers | `upDownCounter` | `integer` | - | - | +| active_readers | `upDownCounter` | `integer` | - | - | +| read_bytes | `counter` | `integer` | - | - | + +| Name | Type | +|---------|---------| diff --git a/ingress/metrics/file_stage.metrics.doc.md b/ingress/metrics/file_stage.metrics.doc.md new file mode 100644 index 0000000..9ab41a0 --- /dev/null +++ b/ingress/metrics/file_stage.metrics.doc.md @@ -0,0 +1,5 @@ +| Name | Type | Attributes | Description | +|---------|---------|---------|---------| +| readers | `upDownCounter` `integer` | - | - | +| active_readers | `upDownCounter` `integer` | - | - | +| read_bytes | `counter` `integer` | - | - | diff --git a/ingress/metrics/http_stage.attributes.doc.md b/ingress/metrics/http_stage.attributes.doc.md new file mode 100644 index 0000000..1ca067d --- /dev/null +++ b/ingress/metrics/http_stage.attributes.doc.md @@ -0,0 +1,7 @@ +| Name | Type | Description | +|---------|---------|---------| +| `http.response.status_code` {#http.response.status_code} | `int` | - | +| `network.protocol.version` {#network.protocol.version} | `string` | - | +| `outcome` {#outcome} | `string` | - | +| `http.request.method` {#http.request.method} | `string` | - | +| `url.scheme` {#url.scheme} | `string` | - | diff --git a/ingress/metrics/http_stage.doc.md b/ingress/metrics/http_stage.doc.md index 696b01b..def6691 100644 --- a/ingress/metrics/http_stage.doc.md +++ b/ingress/metrics/http_stage.doc.md @@ -1,10 +1,18 @@ -| Name | Type | Data Type | Description | -|---------|---------|---------|---------| -| http.server.request.duration | `histogram` | `float` | Duration of HTTP server requests from handler entry until the response is written. | -| http.server.active_requests | `upDownCounter` | `integer` | Number of HTTP server requests currently being handled. | -| http.server.request.body.size | `histogram` | `integer` | Size of HTTP request bodies observed by the server, including partially read bodies. | -| http.server.response.body.size | `histogram` | `integer` | Size of HTTP response bodies successfully written by the server. | -| goccia.http.ingress.queue.len | `gauge` | `integer` | Number of HTTP request messages currently waiting in the ingress output queue. | -| goccia.http.ingress.queue.wait.duration | `histogram` | `float` | Time spent waiting to enqueue an HTTP request message for downstream processing. | -| goccia.http.ingress.pending_responses | `upDownCounter` | `integer` | Number of HTTP requests currently awaiting a downstream response. | -| goccia.http.ingress.response.wait.duration | `histogram` | `float` | Time spent waiting for a downstream response after a successful queue handoff. | +| Name | Type | Data Type | Attributes | Description | +|---------|---------|---------|---------|---------| +| http.server.request.duration | `histogram` | `float` | [http.request.method](#http.request.method), [url.scheme](#url.scheme), [http.response.status_code](#http.response.status_code), [network.protocol.version](#network.protocol.version) | Duration of HTTP server requests from handler entry until the response is written. | +| http.server.active_requests | `upDownCounter` | `integer` | - | Number of HTTP server requests currently being handled. | +| http.server.request.body.size | `histogram` | `integer` | [http.request.method](#http.request.method), [url.scheme](#url.scheme), [http.response.status_code](#http.response.status_code), [network.protocol.version](#network.protocol.version) | Size of HTTP request bodies observed by the server, including partially read bodies. | +| http.server.response.body.size | `histogram` | `integer` | [http.request.method](#http.request.method), [url.scheme](#url.scheme), [http.response.status_code](#http.response.status_code), [network.protocol.version](#network.protocol.version) | Size of HTTP response bodies successfully written by the server. | +| goccia.http.ingress.queue.len | `gauge` | `integer` | - | Number of HTTP request messages currently waiting in the ingress output queue. | +| goccia.http.ingress.queue.wait.duration | `histogram` | `float` | [outcome](#outcome) | Time spent waiting to enqueue an HTTP request message for downstream processing. | +| goccia.http.ingress.pending_responses | `upDownCounter` | `integer` | - | Number of HTTP requests currently awaiting a downstream response. | +| goccia.http.ingress.response.wait.duration | `histogram` | `float` | [outcome](#outcome) | Time spent waiting for a downstream response after a successful queue handoff. | + +| Name | Type | +|---------|---------| +| `http.response.status_code` {#http.response.status_code} | int | +| `network.protocol.version` {#network.protocol.version} | string | +| `outcome` {#outcome} | string | +| `http.request.method` {#http.request.method} | string | +| `url.scheme` {#url.scheme} | string | diff --git a/ingress/metrics/http_stage.metrics.doc.md b/ingress/metrics/http_stage.metrics.doc.md new file mode 100644 index 0000000..5c24e4f --- /dev/null +++ b/ingress/metrics/http_stage.metrics.doc.md @@ -0,0 +1,10 @@ +| Name | Type | Attributes | Description | +|---------|---------|---------|---------| +| http.server.request.duration | `histogram` `float` | [http.request.method](#http.request.method) [url.scheme](#url.scheme) [http.response.status_code](#http.response.status_code) [network.protocol.version](#network.protocol.version) | Duration of HTTP server requests from handler entry until the response is written. | +| http.server.active_requests | `upDownCounter` `integer` | - | Number of HTTP server requests currently being handled. | +| http.server.request.body.size | `histogram` `integer` | [http.request.method](#http.request.method) [url.scheme](#url.scheme) [http.response.status_code](#http.response.status_code) [network.protocol.version](#network.protocol.version) | Size of HTTP request bodies observed by the server, including partially read bodies. | +| http.server.response.body.size | `histogram` `integer` | [http.request.method](#http.request.method) [url.scheme](#url.scheme) [http.response.status_code](#http.response.status_code) [network.protocol.version](#network.protocol.version) | Size of HTTP response bodies successfully written by the server. | +| goccia.http.ingress.queue.len | `gauge` `integer` | - | Number of HTTP request messages currently waiting in the ingress output queue. | +| goccia.http.ingress.queue.wait.duration | `histogram` `float` | [outcome](#outcome) | Time spent waiting to enqueue an HTTP request message for downstream processing. | +| goccia.http.ingress.pending_responses | `upDownCounter` `integer` | - | Number of HTTP requests currently awaiting a downstream response. | +| goccia.http.ingress.response.wait.duration | `histogram` `float` | [outcome](#outcome) | Time spent waiting for a downstream response after a successful queue handoff. | diff --git a/ingress/metrics/http_stage.metrics.go b/ingress/metrics/http_stage.metrics.go index 2ae8794..ca01694 100644 --- a/ingress/metrics/http_stage.metrics.go +++ b/ingress/metrics/http_stage.metrics.go @@ -143,7 +143,7 @@ func (m *HttpStage) RecordHttpServerRequestDuration( } // RecordHttpServerRequestDurationWithAttributes records the given value -// ans attributes into the histogram metric. +// and attributes into the histogram metric. // // HttpServerRequestDuration: Duration of HTTP server requests from handler entry until the response is written. func (m *HttpStage) RecordHttpServerRequestDurationWithAttributes( @@ -202,7 +202,7 @@ func (m *HttpStage) RecordHttpServerRequestBodySize( } // RecordHttpServerRequestBodySizeWithAttributes records the given value -// ans attributes into the histogram metric. +// and attributes into the histogram metric. // // HttpServerRequestBodySize: Size of HTTP request bodies observed by the server, including partially read bodies. func (m *HttpStage) RecordHttpServerRequestBodySizeWithAttributes( @@ -239,7 +239,7 @@ func (m *HttpStage) RecordHttpServerResponseBodySize( } // RecordHttpServerResponseBodySizeWithAttributes records the given value -// ans attributes into the histogram metric. +// and attributes into the histogram metric. // // HttpServerResponseBodySize: Size of HTTP response bodies successfully written by the server. func (m *HttpStage) RecordHttpServerResponseBodySizeWithAttributes( @@ -292,7 +292,7 @@ func (m *HttpStage) RecordGocciaHttpIngressQueueWaitDuration( } // RecordGocciaHttpIngressQueueWaitDurationWithAttributes records the given value -// ans attributes into the histogram metric. +// and attributes into the histogram metric. // // GocciaHttpIngressQueueWaitDuration: Time spent waiting to enqueue an HTTP request message for downstream processing. func (m *HttpStage) RecordGocciaHttpIngressQueueWaitDurationWithAttributes( @@ -345,7 +345,7 @@ func (m *HttpStage) RecordGocciaHttpIngressResponseWaitDuration( } // RecordGocciaHttpIngressResponseWaitDurationWithAttributes records the given value -// ans attributes into the histogram metric. +// and attributes into the histogram metric. // // GocciaHttpIngressResponseWaitDuration: Time spent waiting for a downstream response after a successful queue handoff. func (m *HttpStage) RecordGocciaHttpIngressResponseWaitDurationWithAttributes( diff --git a/ingress/metrics/kafka_stage.doc.md b/ingress/metrics/kafka_stage.doc.md index 20c8ba5..a6127ad 100644 --- a/ingress/metrics/kafka_stage.doc.md +++ b/ingress/metrics/kafka_stage.doc.md @@ -1,4 +1,7 @@ -| Name | Type | Data Type | Description | -|---------|---------|---------|---------| -| received_messages | `counter` | `integer` | - | -| received_bytes | `counter` | `integer` | - | +| Name | Type | Data Type | Attributes | Description | +|---------|---------|---------|---------|---------| +| received_messages | `counter` | `integer` | - | - | +| received_bytes | `counter` | `integer` | - | - | + +| Name | Type | +|---------|---------| diff --git a/ingress/metrics/kafka_stage.metrics.doc.md b/ingress/metrics/kafka_stage.metrics.doc.md new file mode 100644 index 0000000..6e0a529 --- /dev/null +++ b/ingress/metrics/kafka_stage.metrics.doc.md @@ -0,0 +1,4 @@ +| Name | Type | Attributes | Description | +|---------|---------|---------|---------| +| received_messages | `counter` `integer` | - | - | +| received_bytes | `counter` `integer` | - | - | diff --git a/ingress/metrics/spec.yaml b/ingress/metrics/spec.yaml index c2ce813..76cce81 100644 --- a/ingress/metrics/spec.yaml +++ b/ingress/metrics/spec.yaml @@ -5,15 +5,19 @@ attribute_sets: attributes: - name: http.request.method type: string + description: Request method. arg: requestMethod - name: url.scheme type: string + description: "`http` or `https`, inferred from the request's TLS state." arg: urlScheme - name: http.response.status_code type: int + description: Status recorded by the response writer. arg: statusCode - name: network.protocol.version type: string + description: "`1.0`, `1.1`, `2`, or another major/minor version reported by `net/http`." arg: protocolVersion groups: diff --git a/ingress/metrics/tcp_stage.doc.md b/ingress/metrics/tcp_stage.doc.md index f99faa5..ee6dd41 100644 --- a/ingress/metrics/tcp_stage.doc.md +++ b/ingress/metrics/tcp_stage.doc.md @@ -1,5 +1,8 @@ -| Name | Type | Data Type | Description | -|---------|---------|---------|---------| -| open_connections | `upDownCounter` | `integer` | - | -| received_messages | `counter` | `integer` | - | -| received_bytes | `counter` | `integer` | - | +| Name | Type | Data Type | Attributes | Description | +|---------|---------|---------|---------|---------| +| open_connections | `upDownCounter` | `integer` | - | - | +| received_messages | `counter` | `integer` | - | - | +| received_bytes | `counter` | `integer` | - | - | + +| Name | Type | +|---------|---------| diff --git a/ingress/metrics/tcp_stage.metrics.doc.md b/ingress/metrics/tcp_stage.metrics.doc.md new file mode 100644 index 0000000..a6a264e --- /dev/null +++ b/ingress/metrics/tcp_stage.metrics.doc.md @@ -0,0 +1,5 @@ +| Name | Type | Attributes | Description | +|---------|---------|---------|---------| +| open_connections | `upDownCounter` `integer` | - | - | +| received_messages | `counter` `integer` | - | - | +| received_bytes | `counter` `integer` | - | - | diff --git a/ingress/metrics/ticker_stage.doc.md b/ingress/metrics/ticker_stage.doc.md index 5e109a5..83effd9 100644 --- a/ingress/metrics/ticker_stage.doc.md +++ b/ingress/metrics/ticker_stage.doc.md @@ -1,3 +1,6 @@ -| Name | Type | Data Type | Description | -|---------|---------|---------|---------| -| triggered_messages | `counter` | `integer` | - | +| Name | Type | Data Type | Attributes | Description | +|---------|---------|---------|---------|---------| +| triggered_messages | `counter` | `integer` | - | - | + +| Name | Type | +|---------|---------| diff --git a/ingress/metrics/ticker_stage.metrics.doc.md b/ingress/metrics/ticker_stage.metrics.doc.md new file mode 100644 index 0000000..b462480 --- /dev/null +++ b/ingress/metrics/ticker_stage.metrics.doc.md @@ -0,0 +1,3 @@ +| Name | Type | Attributes | Description | +|---------|---------|---------|---------| +| triggered_messages | `counter` `integer` | - | - | diff --git a/ingress/metrics/udp_stage.doc.md b/ingress/metrics/udp_stage.doc.md index 20c8ba5..a6127ad 100644 --- a/ingress/metrics/udp_stage.doc.md +++ b/ingress/metrics/udp_stage.doc.md @@ -1,4 +1,7 @@ -| Name | Type | Data Type | Description | -|---------|---------|---------|---------| -| received_messages | `counter` | `integer` | - | -| received_bytes | `counter` | `integer` | - | +| Name | Type | Data Type | Attributes | Description | +|---------|---------|---------|---------|---------| +| received_messages | `counter` | `integer` | - | - | +| received_bytes | `counter` | `integer` | - | - | + +| Name | Type | +|---------|---------| diff --git a/ingress/metrics/udp_stage.metrics.doc.md b/ingress/metrics/udp_stage.metrics.doc.md new file mode 100644 index 0000000..6e0a529 --- /dev/null +++ b/ingress/metrics/udp_stage.metrics.doc.md @@ -0,0 +1,4 @@ +| Name | Type | Attributes | Description | +|---------|---------|---------|---------| +| received_messages | `counter` `integer` | - | - | +| received_bytes | `counter` `integer` | - | - | diff --git a/internal/stage/metrics/docs/egress_stage.metrics.doc.md b/internal/stage/metrics/docs/egress_stage.metrics.doc.md new file mode 100644 index 0000000..d5058b6 --- /dev/null +++ b/internal/stage/metrics/docs/egress_stage.metrics.doc.md @@ -0,0 +1,5 @@ +| Name | Type | Attributes | Description | +|---------|---------|---------|---------| +| delivered_messages | `counter` `integer` | - | - | +| delivering_errors | `counter` `integer` | - | - | +| total_message_processing_time | `histogram` `integer` | - | - | diff --git a/internal/stage/metrics/docs/processor_stage.metrics.doc.md b/internal/stage/metrics/docs/processor_stage.metrics.doc.md new file mode 100644 index 0000000..7c166c4 --- /dev/null +++ b/internal/stage/metrics/docs/processor_stage.metrics.doc.md @@ -0,0 +1,5 @@ +| Name | Type | Attributes | Description | +|---------|---------|---------|---------| +| processed_messages | `counter` `integer` | - | - | +| dropped_messages | `counter` `integer` | - | - | +| processing_errors | `counter` `integer` | - | - | diff --git a/internal/stage/metrics/egress_stage.doc.md b/internal/stage/metrics/egress_stage.doc.md index f5860fa..4161076 100644 --- a/internal/stage/metrics/egress_stage.doc.md +++ b/internal/stage/metrics/egress_stage.doc.md @@ -1,5 +1,8 @@ -| Name | Type | Data Type | Description | -|---------|---------|---------|---------| -| delivered_messages | `counter` | `integer` | - | -| delivering_errors | `counter` | `integer` | - | -| total_message_processing_time | `histogram` | `integer` | - | +| Name | Type | Data Type | Attributes | Description | +|---------|---------|---------|---------|---------| +| delivered_messages | `counter` | `integer` | - | - | +| delivering_errors | `counter` | `integer` | - | - | +| total_message_processing_time | `histogram` | `integer` | - | - | + +| Name | Type | +|---------|---------| diff --git a/internal/stage/metrics/egress_stage.metrics.doc.md b/internal/stage/metrics/egress_stage.metrics.doc.md new file mode 100644 index 0000000..d5058b6 --- /dev/null +++ b/internal/stage/metrics/egress_stage.metrics.doc.md @@ -0,0 +1,5 @@ +| Name | Type | Attributes | Description | +|---------|---------|---------|---------| +| delivered_messages | `counter` `integer` | - | - | +| delivering_errors | `counter` `integer` | - | - | +| total_message_processing_time | `histogram` `integer` | - | - | diff --git a/internal/stage/metrics/egress_stage.metrics.go b/internal/stage/metrics/egress_stage.metrics.go index e993c4f..13d34ed 100644 --- a/internal/stage/metrics/egress_stage.metrics.go +++ b/internal/stage/metrics/egress_stage.metrics.go @@ -102,7 +102,7 @@ func (m *EgressStage) RecordTotalMessageProcessingTime( } // RecordTotalMessageProcessingTimeWithAttributes records the given value -// ans attributes into the histogram metric. +// and attributes into the histogram metric. func (m *EgressStage) RecordTotalMessageProcessingTimeWithAttributes( ctx context.Context, value int64, diff --git a/internal/stage/metrics/processor_stage.doc.md b/internal/stage/metrics/processor_stage.doc.md index 342890d..9399be0 100644 --- a/internal/stage/metrics/processor_stage.doc.md +++ b/internal/stage/metrics/processor_stage.doc.md @@ -1,5 +1,8 @@ -| Name | Type | Data Type | Description | -|---------|---------|---------|---------| -| processed_messages | `counter` | `integer` | - | -| dropped_messages | `counter` | `integer` | - | -| processing_errors | `counter` | `integer` | - | +| Name | Type | Data Type | Attributes | Description | +|---------|---------|---------|---------|---------| +| processed_messages | `counter` | `integer` | - | - | +| dropped_messages | `counter` | `integer` | - | - | +| processing_errors | `counter` | `integer` | - | - | + +| Name | Type | +|---------|---------| diff --git a/internal/stage/metrics/processor_stage.metrics.doc.md b/internal/stage/metrics/processor_stage.metrics.doc.md new file mode 100644 index 0000000..7c166c4 --- /dev/null +++ b/internal/stage/metrics/processor_stage.metrics.doc.md @@ -0,0 +1,5 @@ +| Name | Type | Attributes | Description | +|---------|---------|---------|---------| +| processed_messages | `counter` `integer` | - | - | +| dropped_messages | `counter` `integer` | - | - | +| processing_errors | `counter` `integer` | - | - | diff --git a/processor/json.go b/processor/json.go new file mode 100644 index 0000000..fd8e4f6 --- /dev/null +++ b/processor/json.go @@ -0,0 +1,20 @@ +package processor + +// ─── Message ────────────────────────────────────────────────────────────────| + +// JSONMessage carries a typed value between JSON processor stages. +type JSONMessage[T any] struct { + // Data is the decoded value or the value to encode. + Data T +} + +// NewJSONMessage returns a new JSONMessage with the given generic data. +func NewJSONMessage[T any](data T) *JSONMessage[T] { + return &JSONMessage[T]{ + Data: data, + } +} + +// Destroy releases resources owned by the message. JSONMessage owns no +// external resources, so Destroy is a no-op. +func (m *JSONMessage[T]) Destroy() {} diff --git a/processor/json_decoder.go b/processor/json_decoder.go new file mode 100644 index 0000000..1b8366e --- /dev/null +++ b/processor/json_decoder.go @@ -0,0 +1,344 @@ +package processor + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "strings" + "time" + + "github.com/FerroO2000/goccia/internal/config" + "github.com/FerroO2000/goccia/internal/message" + "github.com/FerroO2000/goccia/internal/stage" + "github.com/FerroO2000/goccia/internal/stage/env" + "github.com/FerroO2000/goccia/internal/stage/worker" + "github.com/FerroO2000/goccia/processor/metrics" +) + +// ─── Errors ─────────────────────────────────────────────────────────────────| + +var ( + // ErrJSONInputTooLarge is returned when the input size exceeds the maximum + // allowed size. + ErrJSONInputTooLarge = errors.New("JSON input exceeds maximum size") + + // ErrJSONNullRejected is returned when a top-level JSON null value is + // rejected. + ErrJSONNullRejected = errors.New("top-level JSON null is not allowed") + + // ErrJSONTrailingValue is returned when the input contains multiple top-level + // values. + ErrJSONTrailingValue = errors.New("JSON input contains multiple top-level values") +) + +// ─── Config ─────────────────────────────────────────────────────────────────| + +const ( + // DefaultJSONDecoderConfigDisallowUnknownFields permits JSON object members + // that do not match fields in a struct destination. + DefaultJSONDecoderConfigDisallowUnknownFields = false + + // DefaultJSONDecoderConfigUseNumber decodes numbers stored in interface + // values as float64 values. + DefaultJSONDecoderConfigUseNumber = false + + // DefaultJSONDecoderConfigMaxInputBytes disables the decoder input-size + // limit. + DefaultJSONDecoderConfigMaxInputBytes = 0 + + // DefaultJSONDecoderConfigRejectNull permits a top-level JSON null value. + DefaultJSONDecoderConfigRejectNull = false +) + +// JSONDecoderConfig configures a [JSONDecoderStage]. Decoder-specific settings +// are captured when the stage is initialized. +type JSONDecoderConfig struct { + *config.Base + + // DisallowUnknownFields rejects object keys that do not match an exported, + // non-ignored field when an object is decoded into a struct. Objects decoded + // into map or interface values are unaffected. + DisallowUnknownFields bool + + // UseNumber decodes numbers stored in interface values as [json.Number] + // instead of float64. Typed numeric fields are unaffected. + UseNumber bool + + // MaxInputBytes is the maximum accepted raw input size in bytes, including + // leading and trailing whitespace. + // A value of zero disables the limit. + // Negative values are reset to zero during validation. + MaxInputBytes int + + // RejectNull rejects a top-level JSON null value. Nested null values are + // unaffected. + RejectNull bool +} + +// NewJSONDecoderConfig returns the default JSON decoder configuration. +func NewJSONDecoderConfig(runningMode config.StageRunningMode) *JSONDecoderConfig { + return &JSONDecoderConfig{ + Base: config.NewBase(runningMode), + + DisallowUnknownFields: DefaultJSONDecoderConfigDisallowUnknownFields, + UseNumber: DefaultJSONDecoderConfigUseNumber, + MaxInputBytes: DefaultJSONDecoderConfigMaxInputBytes, + RejectNull: DefaultJSONDecoderConfigRejectNull, + } +} + +// Validate checks the JSON decoder configuration. +func (c *JSONDecoderConfig) Validate(ac *config.AnomalyCollector) { + c.Base.Validate(ac) + + config.CheckNotNegative( + ac, "MaxInputBytes", &c.MaxInputBytes, DefaultJSONDecoderConfigMaxInputBytes, + ) +} + +// ─── Decoder ────────────────────────────────────────────────────────────────| + +type jsonDecoderConfig struct { + disallowUnknownFields bool + useNumber bool + maxInputBytes int + rejectNull bool +} + +type jsonDecoderMode int8 + +const ( + jsonDecoderModeDefault jsonDecoderMode = iota + jsonDecoderModeConfigured +) + +func (dm jsonDecoderMode) fromConfig(config jsonDecoderConfig) jsonDecoderMode { + if config.disallowUnknownFields || config.useNumber { + return jsonDecoderModeConfigured + } + + return jsonDecoderModeDefault +} + +type jsonDecoder[T any] struct { + config jsonDecoderConfig + + mode jsonDecoderMode +} + +func newJSONDecoder[T any](config jsonDecoderConfig) *jsonDecoder[T] { + return &jsonDecoder[T]{ + config: config, + + mode: jsonDecoderModeDefault.fromConfig(config), + } +} + +func (d *jsonDecoder[T]) checkDataSize(data []byte) error { + if len(data) > d.config.maxInputBytes { + return fmt.Errorf( + "%w: got %d bytes, maximum is %d bytes", + ErrJSONInputTooLarge, len(data), d.config.maxInputBytes, + ) + } + + return nil +} + +func (d *jsonDecoder[T]) decode(data []byte) (T, error) { + var res T + + if d.config.maxInputBytes > 0 { + if err := d.checkDataSize(data); err != nil { + return res, err + } + } + + var err error + switch d.mode { + case jsonDecoderModeConfigured: + err = d.decodeConfigured(data, &res) + + default: + err = json.Unmarshal(data, &res) + } + if err != nil { + return res, err + } + + if d.config.rejectNull && d.isJSONNull(data) { + return res, ErrJSONNullRejected + } + + return res, nil +} + +func (d *jsonDecoder[T]) decodeConfigured(data []byte, res *T) error { + dec := json.NewDecoder(bytes.NewReader(data)) + + if d.config.disallowUnknownFields { + dec.DisallowUnknownFields() + } + + if d.config.useNumber { + dec.UseNumber() + } + + if err := dec.Decode(res); err != nil { + return err + } + + var trailing json.RawMessage + if err := dec.Decode(&trailing); err != io.EOF { + if err != nil { + return err + } + + return ErrJSONTrailingValue + } + + return nil +} + +func (d *jsonDecoder[T]) isJSONNull(data []byte) bool { + return bytes.Equal(bytes.Trim(data, " \t\r\n"), []byte("null")) +} + +// ─── Environment ────────────────────────────────────────────────────────────| + +type jsonDecoderEnv[T any] struct { + *env.BaseEnv[*JSONDecoderConfig, *metrics.JsonDecoder] + + decoder *jsonDecoder[T] +} + +func newJSONDecoderEnv[T any](config *JSONDecoderConfig) *jsonDecoderEnv[T] { + return &jsonDecoderEnv[T]{ + BaseEnv: env.NewProcessorEnv(config, metrics.NewJsonDecoder()), + } +} + +func (e *jsonDecoderEnv[T]) Init(ctx context.Context) error { + if err := e.BaseEnv.Init(ctx); err != nil { + return err + } + + e.decoder = newJSONDecoder[T](jsonDecoderConfig{ + disallowUnknownFields: e.Config.DisallowUnknownFields, + useNumber: e.Config.UseNumber, + maxInputBytes: e.Config.MaxInputBytes, + rejectNull: e.Config.RejectNull, + }) + + return nil +} + +// ─── Worker ─────────────────────────────────────────────────────────────────| + +type jsonDecoderWorker[In msgSer, Out any] struct { + worker.BaseWorker[*jsonDecoderEnv[Out]] +} + +func newJSONDecoderWorkerMaker[In msgSer, Out any]() func() *jsonDecoderWorker[In, Out] { + return func() *jsonDecoderWorker[In, Out] { + return &jsonDecoderWorker[In, Out]{} + } +} + +func (w *jsonDecoderWorker[In, Out]) getErrorType(err error) metrics.JsonDecoderErrorType { + switch { + case errors.Is(err, ErrJSONInputTooLarge): + return metrics.JsonDecoderErrorTypeInputTooLarge + case errors.Is(err, ErrJSONNullRejected): + return metrics.JsonDecoderErrorTypeNullRejected + case errors.Is(err, ErrJSONTrailingValue): + return metrics.JsonDecoderErrorTypeTrailingValue + } + + var syntaxErr *json.SyntaxError + if errors.As(err, &syntaxErr) || errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) { + return metrics.JsonDecoderErrorTypeSyntaxError + } + + var unmarshalTypeErr *json.UnmarshalTypeError + if errors.As(err, &unmarshalTypeErr) { + return metrics.JsonDecoderErrorTypeTypeError + } + + // DisallowUnknownFields returns a formatted error without an exported, + // distinguishable error type in encoding/json. + if strings.HasPrefix(err.Error(), "json: unknown field ") { + return metrics.JsonDecoderErrorTypeUnknownField + } + + return metrics.JsonDecoderErrorTypeOther +} + +func (w *jsonDecoderWorker[In, Out]) handleMetrics( + ctx context.Context, decDuration float64, inputSize int64, err error) { + + if err != nil { + errType := w.getErrorType(err) + w.Env.Metrics.RecordGocciaJsonDecoderOperationDurationWithErrorType(ctx, decDuration, errType) + w.Env.Metrics.RecordGocciaJsonDecoderInputSizeWithErrorType(ctx, inputSize, errType) + return + } + + w.Env.Metrics.RecordGocciaJsonDecoderOperationDuration(ctx, decDuration) + w.Env.Metrics.RecordGocciaJsonDecoderInputSize(ctx, inputSize) +} + +func (w *jsonDecoderWorker[In, Out]) Handle(ctx context.Context, msgIn *msg[In]) (*msg[*JSONMessage[Out]], error) { + _, span := w.Tel.StartTrace(ctx, "decode json data") + defer span.End() + + decStartTime := time.Now() + + inputData := msgIn.GetBody().GetBytes() + inputSize := len(inputData) + + decodedData, err := w.Env.decoder.decode(inputData) + + decDuration := time.Since(decStartTime).Seconds() + w.handleMetrics(ctx, decDuration, int64(inputSize), err) + + if err != nil { + return nil, err + } + + jsonMsg := NewJSONMessage(decodedData) + msgOut := message.NewMessage(jsonMsg) + + msgOut.SaveSpan(span) + + return msgOut, nil +} + +// ─── Stage ──────────────────────────────────────────────────────────────────| + +var _ stage.Stage = (*JSONDecoderStage[msgSer, *any])(nil) + +// JSONDecoderStage decodes serialized JSON into values of type Out. Out can be +// any type supported by encoding/json. When Out is a pointer type and RejectNull +// is disabled, a top-level JSON null value produces a nil Out value. +type JSONDecoderStage[In msgSer, Out any] struct { + *stage.ProcessorStage[In, *JSONMessage[Out], *jsonDecoderEnv[Out]] +} + +// NewJSONDecoderStage returns a JSON decoder stage that reads from inConnector +// and writes decoded messages to outConnector. Config must be non-nil. +func NewJSONDecoderStage[In msgSer, Out any]( + inConnector msgConn[In], outConnector msgConn[*JSONMessage[Out]], config *JSONDecoderConfig, +) *JSONDecoderStage[In, Out] { + + env := newJSONDecoderEnv[Out](config) + + return &JSONDecoderStage[In, Out]{ + ProcessorStage: stage.NewProcessorStage( + "json_decoder", inConnector, outConnector, env, newJSONDecoderWorkerMaker[In, Out](), config.Stage, + ), + } +} diff --git a/processor/json_decoder_test.go b/processor/json_decoder_test.go new file mode 100644 index 0000000..8fc4d6f --- /dev/null +++ b/processor/json_decoder_test.go @@ -0,0 +1,338 @@ +package processor + +import ( + "encoding/json" + "testing" + + "github.com/stretchr/testify/assert" +) + +type jsonDecoderTestPayload struct { + Name string `json:"name"` + Count int `json:"count"` + Number any `json:"number"` + Child *jsonDecoderTestPayload `json:"child"` +} + +type jsonDecoderTestCustom struct { + Value string +} + +func (c *jsonDecoderTestCustom) UnmarshalJSON(data []byte) error { + return json.Unmarshal(data, &c.Value) +} + +func Test_jsonDecoder_New(t *testing.T) { + tests := []struct { + name string + config jsonDecoderConfig + mode jsonDecoderMode + }{ + { + name: "default", + mode: jsonDecoderModeDefault, + }, + { + name: "input size limit", + config: jsonDecoderConfig{maxInputBytes: 1}, + mode: jsonDecoderModeDefault, + }, + { + name: "reject null", + config: jsonDecoderConfig{rejectNull: true}, + mode: jsonDecoderModeDefault, + }, + { + name: "disallow unknown fields", + config: jsonDecoderConfig{disallowUnknownFields: true}, + mode: jsonDecoderModeConfigured, + }, + { + name: "use number", + config: jsonDecoderConfig{useNumber: true}, + mode: jsonDecoderModeConfigured, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert := assert.New(t) + + decoder := newJSONDecoder[*jsonDecoderTestPayload](tt.config) + + assert.Equal(tt.config, decoder.config) + assert.Equal(tt.mode, decoder.mode) + }) + } +} + +func Test_jsonDecoder_Decode(t *testing.T) { + t.Run("pointer target", func(t *testing.T) { + assert := assert.New(t) + decoder := newJSONDecoder[*jsonDecoderTestPayload](jsonDecoderConfig{}) + + got, err := decoder.decode([]byte(`{"name":"goccia","count":1}`)) + if !assert.NoError(err) || !assert.NotNil(got) { + return + } + + assert.Equal("goccia", got.Name) + assert.Equal(1, got.Count) + }) + + t.Run("pointer unmarshaler", func(t *testing.T) { + assert := assert.New(t) + decoder := newJSONDecoder[*jsonDecoderTestCustom](jsonDecoderConfig{}) + + got, err := decoder.decode([]byte(`"custom"`)) + if !assert.NoError(err) || !assert.NotNil(got) { + return + } + + assert.Equal("custom", got.Value) + }) + + t.Run("null", testJSONDecoderNull) + t.Run("unknown fields", testJSONDecoderUnknownFields) + t.Run("numbers", testJSONDecoderNumbers) + t.Run("input size", testJSONDecoderInputSize) + t.Run("trailing data", testJSONDecoderTrailingData) + t.Run("malformed input", testJSONDecoderMalformedInput) +} + +func testJSONDecoderNull(t *testing.T) { + t.Run("allowed", func(t *testing.T) { + assert := assert.New(t) + decoder := newJSONDecoder[*jsonDecoderTestPayload](jsonDecoderConfig{}) + + got, err := decoder.decode([]byte("null")) + + assert.NoError(err) + assert.Nil(got) + }) + + rejectingConfigs := []struct { + name string + config jsonDecoderConfig + }{ + { + name: "unmarshal", + config: jsonDecoderConfig{rejectNull: true}, + }, + { + name: "configured decoder", + config: jsonDecoderConfig{ + rejectNull: true, + useNumber: true, + }, + }, + } + + for _, tt := range rejectingConfigs { + t.Run("rejected "+tt.name, func(t *testing.T) { + assert := assert.New(t) + decoder := newJSONDecoder[*jsonDecoderTestPayload](tt.config) + + for _, input := range [][]byte{[]byte("null"), []byte(" \nnull\t\r")} { + _, err := decoder.decode(input) + assert.ErrorIs(err, ErrJSONNullRejected) + } + }) + } + + t.Run("nested value allowed", func(t *testing.T) { + assert := assert.New(t) + decoder := newJSONDecoder[*jsonDecoderTestPayload](jsonDecoderConfig{rejectNull: true}) + + got, err := decoder.decode([]byte(`{"name":"parent","child":null}`)) + if !assert.NoError(err) || !assert.NotNil(got) { + return + } + + assert.Nil(got.Child) + }) +} + +func testJSONDecoderUnknownFields(t *testing.T) { + data := []byte(`{"name":"goccia","unknown":true}`) + + t.Run("allowed", func(t *testing.T) { + assert := assert.New(t) + decoder := newJSONDecoder[*jsonDecoderTestPayload](jsonDecoderConfig{}) + + got, err := decoder.decode(data) + if !assert.NoError(err) || !assert.NotNil(got) { + return + } + + assert.Equal("goccia", got.Name) + }) + + t.Run("rejected", func(t *testing.T) { + assert := assert.New(t) + decoder := newJSONDecoder[*jsonDecoderTestPayload](jsonDecoderConfig{ + disallowUnknownFields: true, + }) + + _, err := decoder.decode(data) + + if assert.Error(err) { + assert.Contains(err.Error(), "unknown field") + } + }) +} + +func testJSONDecoderNumbers(t *testing.T) { + data := []byte(`{"number":9007199254740993}`) + + t.Run("float64", func(t *testing.T) { + assert := assert.New(t) + decoder := newJSONDecoder[*jsonDecoderTestPayload](jsonDecoderConfig{}) + + got, err := decoder.decode(data) + if !assert.NoError(err) || !assert.NotNil(got) { + return + } + + assert.IsType(float64(0), got.Number) + }) + + t.Run("json number", func(t *testing.T) { + assert := assert.New(t) + decoder := newJSONDecoder[*jsonDecoderTestPayload](jsonDecoderConfig{useNumber: true}) + + got, err := decoder.decode(data) + if !assert.NoError(err) || !assert.NotNil(got) { + return + } + + number, ok := got.Number.(json.Number) + if !assert.True(ok) { + return + } + + assert.Equal("9007199254740993", number.String()) + }) +} + +func testJSONDecoderInputSize(t *testing.T) { + data := []byte(" \n{\"name\":\"goccia\"}\t") + + tests := []struct { + name string + config jsonDecoderConfig + expectsSize bool + }{ + { + name: "unlimited", + config: jsonDecoderConfig{maxInputBytes: 0}, + }, + { + name: "exact limit", + config: jsonDecoderConfig{maxInputBytes: len(data)}, + }, + { + name: "over limit", + config: jsonDecoderConfig{maxInputBytes: len(data) - 1}, + expectsSize: true, + }, + { + name: "configured decoder over limit", + config: jsonDecoderConfig{ + useNumber: true, + maxInputBytes: len(data) - 1, + }, + expectsSize: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert := assert.New(t) + decoder := newJSONDecoder[*jsonDecoderTestPayload](tt.config) + + got, err := decoder.decode(data) + if tt.expectsSize { + assert.ErrorIs(err, ErrJSONInputTooLarge) + assert.Nil(got) + return + } + + if !assert.NoError(err) || !assert.NotNil(got) { + return + } + + assert.Equal("goccia", got.Name) + }) + } +} + +func testJSONDecoderTrailingData(t *testing.T) { + configs := []struct { + name string + config jsonDecoderConfig + }{ + {name: "unmarshal", config: jsonDecoderConfig{}}, + {name: "configured decoder", config: jsonDecoderConfig{useNumber: true}}, + } + + invalidInputs := []struct { + name string + data []byte + }{ + {name: "second object", data: []byte(`{"name":"one"} {"name":"two"}`)}, + {name: "second null", data: []byte(`{"name":"one"}null`)}, + {name: "garbage", data: []byte(`{"name":"one"}x`)}, + } + + for _, cfg := range configs { + t.Run(cfg.name, func(t *testing.T) { + decoder := newJSONDecoder[*jsonDecoderTestPayload](cfg.config) + + for _, input := range invalidInputs { + t.Run(input.name, func(t *testing.T) { + assert := assert.New(t) + + _, err := decoder.decode(input.data) + + assert.Error(err) + }) + } + + t.Run("whitespace", func(t *testing.T) { + assert := assert.New(t) + + got, err := decoder.decode([]byte(" \n{\"name\":\"one\"}\t")) + if !assert.NoError(err) || !assert.NotNil(got) { + return + } + + assert.Equal("one", got.Name) + }) + }) + } +} + +func testJSONDecoderMalformedInput(t *testing.T) { + tests := []struct { + name string + data []byte + }{ + {name: "empty", data: nil}, + {name: "whitespace", data: []byte(" \n\t")}, + {name: "truncated object", data: []byte(`{"name":`)}, + {name: "wrong field type", data: []byte(`{"count":"one"}`)}, + {name: "numeric overflow", data: []byte(`{"count":1e100}`)}, + } + + decoder := newJSONDecoder[*jsonDecoderTestPayload](jsonDecoderConfig{}) + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert := assert.New(t) + + _, err := decoder.decode(tt.data) + + assert.Error(err) + }) + } +} diff --git a/processor/json_encoder.go b/processor/json_encoder.go new file mode 100644 index 0000000..2c35acb --- /dev/null +++ b/processor/json_encoder.go @@ -0,0 +1,283 @@ +package processor + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "time" + + "github.com/FerroO2000/goccia/internal/config" + "github.com/FerroO2000/goccia/internal/message" + "github.com/FerroO2000/goccia/internal/stage" + "github.com/FerroO2000/goccia/internal/stage/env" + "github.com/FerroO2000/goccia/internal/stage/worker" + "github.com/FerroO2000/goccia/processor/metrics" +) + +// ─── Config ─────────────────────────────────────────────────────────────────| + +const ( + // DefaultJSONEncoderConfigEscapeHTML enables escaping of HTML-sensitive + // characters in JSON strings. + DefaultJSONEncoderConfigEscapeHTML = true +) + +// JSONEncoderConfig configures a [JSONEncoderStage]. Encoder-specific settings +// are captured when the stage is initialized. +type JSONEncoderConfig struct { + *config.Base + + // Indent enables pretty printing when non-empty. + // Typical values are " " or "\t". + // It should contain only JSON whitespace to keep the output valid JSON. + Indent string + + // IndentPrefix is written at the beginning of indented lines. + // It is ignored when Indent is empty. + // It should contain only JSON whitespace to keep the output valid JSON. + IndentPrefix string + + // EscapeHTML escapes <, > and & as JSON Unicode sequences. + EscapeHTML bool +} + +// NewJSONEncoderConfig returns the default JSON encoder configuration. +func NewJSONEncoderConfig(runningMode config.StageRunningMode) *JSONEncoderConfig { + return &JSONEncoderConfig{ + Base: config.NewBase(runningMode), + + EscapeHTML: DefaultJSONEncoderConfigEscapeHTML, + } +} + +// Validate checks the JSON encoder configuration. +func (c *JSONEncoderConfig) Validate(ac *config.AnomalyCollector) { + c.Base.Validate(ac) +} + +// ─── Message ────────────────────────────────────────────────────────────────| + +var _ msgSer = (*JSONEncodedMessage)(nil) + +// JSONEncodedMessage contains one JSON-encoded value. +type JSONEncodedMessage struct { + // Data contains the encoded JSON bytes without a trailing newline. + Data []byte +} + +func newJSONEncodedMessage(data []byte) *JSONEncodedMessage { + return &JSONEncodedMessage{ + Data: data, + } +} + +// Destroy releases resources owned by the message. JSONEncodedMessage owns no +// external resources, so Destroy is a no-op. +func (m *JSONEncodedMessage) Destroy() {} + +// GetBytes returns Data without copying it. Callers must not mutate the +// returned slice while the message is in use. +func (m *JSONEncodedMessage) GetBytes() []byte { + return m.Data +} + +// ─── Encoder ────────────────────────────────────────────────────────────────| + +type jsonEncoderConfig struct { + indent string + indentPrefix string + escapeHTML bool +} + +type jsonEncoderMode int8 + +const ( + jsonEncoderModeDefault jsonEncoderMode = iota + jsonEncoderModeIndent + jsonEncoderModeConfigured +) + +func (em jsonEncoderMode) fromConfig(config jsonEncoderConfig) jsonEncoderMode { + if !config.escapeHTML { + return jsonEncoderModeConfigured + } + + if config.indent != "" { + return jsonEncoderModeIndent + } + + return jsonEncoderModeDefault +} + +type jsonEncoder[T any] struct { + config jsonEncoderConfig + + mode jsonEncoderMode +} + +func newJSONEncoder[T any](config jsonEncoderConfig) *jsonEncoder[T] { + return &jsonEncoder[T]{ + config: config, + + mode: jsonEncoderModeDefault.fromConfig(config), + } +} + +func (e *jsonEncoder[T]) encode(dataIn T) ([]byte, error) { + switch e.mode { + case jsonEncoderModeIndent: + return e.encodeWithIndent(dataIn) + + case jsonEncoderModeConfigured: + return e.encodeWithBuffer(dataIn) + + default: + return json.Marshal(dataIn) + } +} + +func (e *jsonEncoder[T]) encodeWithIndent(dataIn T) ([]byte, error) { + return json.MarshalIndent(dataIn, e.config.indentPrefix, e.config.indent) +} + +func (e *jsonEncoder[T]) encodeWithBuffer(dataIn T) ([]byte, error) { + buf := &bytes.Buffer{} + + enc := json.NewEncoder(buf) + enc.SetEscapeHTML(e.config.escapeHTML) + + if e.config.indent != "" { + enc.SetIndent(e.config.indentPrefix, e.config.indent) + } + + if err := enc.Encode(dataIn); err != nil { + return nil, err + } + + // Encode always appends a newline after a successful encoding, + // so it is necessary to trim it + data := buf.Bytes() + + return data[:len(data)-1], nil +} + +// ─── Environment ────────────────────────────────────────────────────────────| + +type jsonEncoderEnv[T any] struct { + *env.BaseEnv[*JSONEncoderConfig, *metrics.JsonEncoder] + + encoder *jsonEncoder[T] +} + +func newJSONEncoderEnv[T any](config *JSONEncoderConfig) *jsonEncoderEnv[T] { + return &jsonEncoderEnv[T]{ + BaseEnv: env.NewProcessorEnv(config, metrics.NewJsonEncoder()), + } +} + +func (e *jsonEncoderEnv[T]) Init(ctx context.Context) error { + if err := e.BaseEnv.Init(ctx); err != nil { + return err + } + + e.encoder = newJSONEncoder[T](jsonEncoderConfig{ + indent: e.Config.Indent, + indentPrefix: e.Config.IndentPrefix, + escapeHTML: e.Config.EscapeHTML, + }) + + return nil +} + +// ─── Worker ─────────────────────────────────────────────────────────────────| + +type jsonEncoderWorker[T any] struct { + worker.BaseWorker[*jsonEncoderEnv[T]] +} + +func newJSONEncoderWorkerMaker[T any]() func() *jsonEncoderWorker[T] { + return func() *jsonEncoderWorker[T] { + return &jsonEncoderWorker[T]{} + } +} + +func (w *jsonEncoderWorker[T]) getErrorType(err error) metrics.JsonEncoderErrorType { + var unsupportedTypeErr *json.UnsupportedTypeError + if errors.As(err, &unsupportedTypeErr) { + return metrics.JsonEncoderErrorTypeUnsupportedType + } + + var unsupportedValueErr *json.UnsupportedValueError + if errors.As(err, &unsupportedValueErr) { + return metrics.JsonEncoderErrorTypeUnsupportedValue + } + + var marshalerErr *json.MarshalerError + if errors.As(err, &marshalerErr) { + return metrics.JsonEncoderErrorTypeMarshalerError + } + + return metrics.JsonEncoderErrorTypeOther +} + +func (w *jsonEncoderWorker[T]) handleMetrics( + ctx context.Context, encDuration float64, outputSize int64, err error) { + + if err != nil { + errType := w.getErrorType(err) + w.Env.Metrics.RecordGocciaJsonEncoderOperationDurationWithErrorType(ctx, encDuration, errType) + return + } + + w.Env.Metrics.RecordGocciaJsonEncoderOperationDuration(ctx, encDuration) + w.Env.Metrics.RecordGocciaJsonEncoderOutputSize(ctx, outputSize) +} + +func (w *jsonEncoderWorker[T]) Handle(ctx context.Context, msgIn *msg[*JSONMessage[T]]) (*msg[*JSONEncodedMessage], error) { + _, span := w.Tel.StartTrace(ctx, "encode json data") + defer span.End() + + encStartTime := time.Now() + + inputData := msgIn.GetBody().Data + data, err := w.Env.encoder.encode(inputData) + + encDuration := time.Since(encStartTime).Seconds() + w.handleMetrics(ctx, encDuration, int64(len(data)), err) + + if err != nil { + return nil, err + } + + jsonEncMsg := newJSONEncodedMessage(data) + msgOut := message.NewMessage(jsonEncMsg) + + msgOut.SaveSpan(span) + + return msgOut, nil +} + +// ─── Stage ──────────────────────────────────────────────────────────────────| + +var _ stage.Stage = (*JSONEncoderStage[any])(nil) + +// JSONEncoderStage encodes the Data field of each [JSONMessage] as JSON. +type JSONEncoderStage[T any] struct { + *stage.ProcessorStage[*JSONMessage[T], *JSONEncodedMessage, *jsonEncoderEnv[T]] +} + +// NewJSONEncoderStage returns a JSON encoder stage that reads from inConnector +// and writes encoded messages to outConnector. Config must be non-nil. +func NewJSONEncoderStage[T any]( + inConnector msgConn[*JSONMessage[T]], outConnector msgConn[*JSONEncodedMessage], config *JSONEncoderConfig, +) *JSONEncoderStage[T] { + + env := newJSONEncoderEnv[T](config) + + return &JSONEncoderStage[T]{ + ProcessorStage: stage.NewProcessorStage( + "json_encoder", inConnector, outConnector, env, newJSONEncoderWorkerMaker[T](), config.Stage, + ), + } +} diff --git a/processor/json_encoder_test.go b/processor/json_encoder_test.go new file mode 100644 index 0000000..8ea72da --- /dev/null +++ b/processor/json_encoder_test.go @@ -0,0 +1,190 @@ +package processor + +import ( + "encoding/json" + "testing" + + "github.com/stretchr/testify/assert" +) + +type jsonEncoderTestPayload struct { + Text string `json:"text"` +} + +type jsonEncoderUnsupportedPayload struct { + Value func() `json:"value"` +} + +func Test_jsonEncoder_New(t *testing.T) { + tests := []struct { + name string + config jsonEncoderConfig + mode jsonEncoderMode + }{ + { + name: "default", + config: jsonEncoderConfig{escapeHTML: true}, + mode: jsonEncoderModeDefault, + }, + { + name: "prefix without indent", + config: jsonEncoderConfig{ + indentPrefix: "\t", + escapeHTML: true, + }, + mode: jsonEncoderModeDefault, + }, + { + name: "indent", + config: jsonEncoderConfig{ + indent: " ", + escapeHTML: true, + }, + mode: jsonEncoderModeIndent, + }, + { + name: "configured", + config: jsonEncoderConfig{escapeHTML: false}, + mode: jsonEncoderModeConfigured, + }, + { + name: "configured with indent", + config: jsonEncoderConfig{ + indent: " ", + escapeHTML: false, + }, + mode: jsonEncoderModeConfigured, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert := assert.New(t) + + encoder := newJSONEncoder[jsonEncoderTestPayload](tt.config) + + assert.Equal(tt.config, encoder.config) + assert.Equal(tt.mode, encoder.mode) + }) + } +} + +func Test_jsonEncoder_Encode(t *testing.T) { + payload := jsonEncoderTestPayload{Text: "&"} + + tests := []struct { + name string + config jsonEncoderConfig + expected string + }{ + { + name: "default", + config: jsonEncoderConfig{escapeHTML: true}, + expected: `{"text":"\u003ctag\u003e\u0026"}`, + }, + { + name: "indent", + config: jsonEncoderConfig{ + indent: " ", + indentPrefix: "\t", + escapeHTML: true, + }, + expected: "{\n" + + "\t \"text\": \"\\u003ctag\\u003e\\u0026\"\n" + + "\t}", + }, + { + name: "configured", + config: jsonEncoderConfig{escapeHTML: false}, + expected: `{"text":"&"}`, + }, + { + name: "configured with indent", + config: jsonEncoderConfig{ + indent: " ", + indentPrefix: "\t", + escapeHTML: false, + }, + expected: "{\n" + + "\t \"text\": \"&\"\n" + + "\t}", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert := assert.New(t) + encoder := newJSONEncoder[jsonEncoderTestPayload](tt.config) + + got, err := encoder.encode(payload) + if !assert.NoError(err) { + return + } + + assert.Equal(tt.expected, string(got)) + if assert.NotEmpty(got) { + assert.NotEqual(byte('\n'), got[len(got)-1]) + } + }) + } +} + +func Test_jsonEncoder_ConfiguredModeRemovesOnlyTerminatingNewline(t *testing.T) { + assert := assert.New(t) + encoder := newJSONEncoder[string](jsonEncoderConfig{escapeHTML: false}) + + got, err := encoder.encode("first\nsecond") + if !assert.NoError(err) { + return + } + + assert.Equal(`"first\nsecond"`, string(got)) + if assert.NotEmpty(got) { + assert.NotEqual(byte('\n'), got[len(got)-1]) + } +} + +func Test_jsonEncoder_EncodeError(t *testing.T) { + tests := []struct { + name string + config jsonEncoderConfig + }{ + { + name: "default", + config: jsonEncoderConfig{escapeHTML: true}, + }, + { + name: "indent", + config: jsonEncoderConfig{ + indent: " ", + escapeHTML: true, + }, + }, + { + name: "configured", + config: jsonEncoderConfig{escapeHTML: false}, + }, + { + name: "configured with indent", + config: jsonEncoderConfig{ + indent: " ", + escapeHTML: false, + }, + }, + } + + payload := jsonEncoderUnsupportedPayload{Value: func() {}} + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert := assert.New(t) + encoder := newJSONEncoder[jsonEncoderUnsupportedPayload](tt.config) + + got, err := encoder.encode(payload) + + assert.Nil(got) + var unsupportedTypeError *json.UnsupportedTypeError + assert.ErrorAs(err, &unsupportedTypeError) + }) + } +} diff --git a/processor/metrics/aggregate_stage.doc.md b/processor/metrics/aggregate_stage.doc.md deleted file mode 100644 index 9028bc2..0000000 --- a/processor/metrics/aggregate_stage.doc.md +++ /dev/null @@ -1,3 +0,0 @@ -| Name | Type | Data Type | Description | -|---------|---------|---------|---------| -| aggregate_messages | `counter` | `integer` | - | diff --git a/processor/metrics/can_stage.doc.md b/processor/metrics/can_stage.doc.md deleted file mode 100644 index f84a7af..0000000 --- a/processor/metrics/can_stage.doc.md +++ /dev/null @@ -1,4 +0,0 @@ -| Name | Type | Data Type | Description | -|---------|---------|---------|---------| -| can_messages | `counter` | `integer` | - | -| can_signals | `counter` | `integer` | - | diff --git a/processor/metrics/docs/aggregate_stage.metrics.doc.md b/processor/metrics/docs/aggregate_stage.metrics.doc.md new file mode 100644 index 0000000..7981d7d --- /dev/null +++ b/processor/metrics/docs/aggregate_stage.metrics.doc.md @@ -0,0 +1,3 @@ +| Name | Type | Attributes | Description | +|---------|---------|---------|---------| +| aggregate_messages | `counter` `integer` | - | - | diff --git a/processor/metrics/docs/can_stage.metrics.doc.md b/processor/metrics/docs/can_stage.metrics.doc.md new file mode 100644 index 0000000..b070ded --- /dev/null +++ b/processor/metrics/docs/can_stage.metrics.doc.md @@ -0,0 +1,4 @@ +| Name | Type | Attributes | Description | +|---------|---------|---------|---------| +| can_messages | `counter` `integer` | - | - | +| can_signals | `counter` `integer` | - | - | diff --git a/processor/metrics/docs/filter_stage.metrics.doc.md b/processor/metrics/docs/filter_stage.metrics.doc.md new file mode 100644 index 0000000..72e60ab --- /dev/null +++ b/processor/metrics/docs/filter_stage.metrics.doc.md @@ -0,0 +1,3 @@ +| Name | Type | Attributes | Description | +|---------|---------|---------|---------| +| filtered_messages | `counter` `integer` | - | - | diff --git a/processor/metrics/docs/json_decoder.error_types.doc.md b/processor/metrics/docs/json_decoder.error_types.doc.md new file mode 100644 index 0000000..a4c1364 --- /dev/null +++ b/processor/metrics/docs/json_decoder.error_types.doc.md @@ -0,0 +1,8 @@ +| Name | Value | Description | +|---------|---------|---------| +| `input_too_large` {#error_type} | `goccia.json.input_too_large` | - | +| null_rejected | `goccia.json.null_rejected` | - | +| trailing_value | `goccia.json.trailing_value` | - | +| syntax_error | `goccia.json.syntax_error` | - | +| type_error | `goccia.json.type_error` | - | +| unknown_field | `goccia.json.unknown_field` | - | diff --git a/processor/metrics/docs/json_decoder.metrics.doc.md b/processor/metrics/docs/json_decoder.metrics.doc.md new file mode 100644 index 0000000..1af3de2 --- /dev/null +++ b/processor/metrics/docs/json_decoder.metrics.doc.md @@ -0,0 +1,4 @@ +| Name | Type | Attributes | Description | +|---------|---------|---------|---------| +| goccia.json.decoder.operation.duration | `histogram` `float` | [error.type](#error_type) | - | +| goccia.json.decoder.input.size | `histogram` `integer` | [error.type](#error_type) | - | diff --git a/processor/metrics/docs/json_encoder.error_types.doc.md b/processor/metrics/docs/json_encoder.error_types.doc.md new file mode 100644 index 0000000..365f038 --- /dev/null +++ b/processor/metrics/docs/json_encoder.error_types.doc.md @@ -0,0 +1,5 @@ +| Name | Value | Description | +|---------|---------|---------| +| `unsupported_type` {#error_type} | `goccia.json.unsupported_type` | - | +| unsupported_value | `goccia.json.unsupported_value` | - | +| marshaler_error | `goccia.json.marshaler_error` | - | diff --git a/processor/metrics/docs/json_encoder.metrics.doc.md b/processor/metrics/docs/json_encoder.metrics.doc.md new file mode 100644 index 0000000..b5402bd --- /dev/null +++ b/processor/metrics/docs/json_encoder.metrics.doc.md @@ -0,0 +1,4 @@ +| Name | Type | Attributes | Description | +|---------|---------|---------|---------| +| goccia.json.encoder.operation.duration | `histogram` `float` | [error.type](#error_type) | - | +| goccia.json.encoder.output.size | `histogram` `integer` | - | - | diff --git a/processor/metrics/docs/rob_stage.metrics.doc.md b/processor/metrics/docs/rob_stage.metrics.doc.md new file mode 100644 index 0000000..dd867aa --- /dev/null +++ b/processor/metrics/docs/rob_stage.metrics.doc.md @@ -0,0 +1,9 @@ +| Name | Type | Attributes | Description | +|---------|---------|---------|---------| +| ordered_messages | `counter` `integer` | - | - | +| primary_enqueued_messages | `counter` `integer` | - | - | +| auxiliary_enqueued_messages | `counter` `integer` | - | - | +| out_of_order_sequence_number | `counter` `integer` | - | - | +| duplicated_sequence_number | `counter` `integer` | - | - | +| invalid_sequence_number | `counter` `integer` | - | - | +| resets | `counter` `integer` | - | - | diff --git a/processor/metrics/docs/router_stage.metrics.doc.md b/processor/metrics/docs/router_stage.metrics.doc.md new file mode 100644 index 0000000..c96585f --- /dev/null +++ b/processor/metrics/docs/router_stage.metrics.doc.md @@ -0,0 +1,3 @@ +| Name | Type | Attributes | Description | +|---------|---------|---------|---------| +| total_routed_messages | `counter` `integer` | - | - | diff --git a/processor/metrics/docs/tee_stage.metrics.doc.md b/processor/metrics/docs/tee_stage.metrics.doc.md new file mode 100644 index 0000000..713e321 --- /dev/null +++ b/processor/metrics/docs/tee_stage.metrics.doc.md @@ -0,0 +1,3 @@ +| Name | Type | Attributes | Description | +|---------|---------|---------|---------| +| cloned_messages | `counter` `integer` | - | - | diff --git a/processor/metrics/error_types.metrics.go b/processor/metrics/error_types.metrics.go new file mode 100644 index 0000000..274d5dd --- /dev/null +++ b/processor/metrics/error_types.metrics.go @@ -0,0 +1,24 @@ +// Code generated by metrics-gen. DO NOT EDIT. + +package metrics + +type JsonDecoderErrorType string + +const ( + JsonDecoderErrorTypeInputTooLarge JsonDecoderErrorType = "goccia.json.input_too_large" + JsonDecoderErrorTypeNullRejected JsonDecoderErrorType = "goccia.json.null_rejected" + JsonDecoderErrorTypeTrailingValue JsonDecoderErrorType = "goccia.json.trailing_value" + JsonDecoderErrorTypeSyntaxError JsonDecoderErrorType = "goccia.json.syntax_error" + JsonDecoderErrorTypeTypeError JsonDecoderErrorType = "goccia.json.type_error" + JsonDecoderErrorTypeUnknownField JsonDecoderErrorType = "goccia.json.unknown_field" + JsonDecoderErrorTypeOther JsonDecoderErrorType = "_OTHER" +) + +type JsonEncoderErrorType string + +const ( + JsonEncoderErrorTypeUnsupportedType JsonEncoderErrorType = "goccia.json.unsupported_type" + JsonEncoderErrorTypeUnsupportedValue JsonEncoderErrorType = "goccia.json.unsupported_value" + JsonEncoderErrorTypeMarshalerError JsonEncoderErrorType = "goccia.json.marshaler_error" + JsonEncoderErrorTypeOther JsonEncoderErrorType = "_OTHER" +) diff --git a/processor/metrics/filter_stage.doc.md b/processor/metrics/filter_stage.doc.md deleted file mode 100644 index bf069a5..0000000 --- a/processor/metrics/filter_stage.doc.md +++ /dev/null @@ -1,3 +0,0 @@ -| Name | Type | Data Type | Description | -|---------|---------|---------|---------| -| filtered_messages | `counter` | `integer` | - | diff --git a/processor/metrics/json_decoder.metrics.go b/processor/metrics/json_decoder.metrics.go new file mode 100644 index 0000000..16e0afa --- /dev/null +++ b/processor/metrics/json_decoder.metrics.go @@ -0,0 +1,154 @@ +// Code generated by metrics-gen. DO NOT EDIT. + +package metrics + +import ( + "context" + "github.com/FerroO2000/goccia/internal/telemetry" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/metric" +) + +// JsonDecoder structs contains the metrics for the json_decoder group. +type JsonDecoder struct { + gocciaJsonDecoderOperationDuration *telemetry.FloatHistogram + gocciaJsonDecoderInputSize *telemetry.IntHistogram + jsonDecoderErrorTypes map[JsonDecoderErrorType]metric.MeasurementOption +} + +// NewJsonDecoder returns a new instance of the JsonDecoder struct. +func NewJsonDecoder() *JsonDecoder { + return &JsonDecoder{} +} + +// InitMetrics initializes the metrics for the JsonDecoder. +// It uses the given telemetry instance to create the metrics. +// +// It must be called before using the metrics. +func (m *JsonDecoder) InitMetrics(tel *telemetry.Telemetry) error { + var err error + + // Initialize goccia.json.decoder.operation.duration metric + m.gocciaJsonDecoderOperationDuration, err = tel.NewFloatHistogramMetric( + "goccia.json.decoder.operation.duration", + metric.WithUnit("s"), + metric.WithExplicitBucketBoundaries( + 1e-06, 2.5e-06, 5e-06, 1e-05, 2.5e-05, 5e-05, 0.0001, 0.00025, 0.0005, 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, + ), + ) + if err != nil { + return err + } + + // Initialize goccia.json.decoder.input.size metric + m.gocciaJsonDecoderInputSize, err = tel.NewIntHistogramMetric( + "goccia.json.decoder.input.size", + metric.WithUnit("By"), + ) + if err != nil { + return err + } + + m.jsonDecoderErrorTypes = map[JsonDecoderErrorType]metric.MeasurementOption{ + JsonDecoderErrorTypeInputTooLarge: metric.WithAttributes( + attribute.String("error.type", string(JsonDecoderErrorTypeInputTooLarge)), + ), + JsonDecoderErrorTypeNullRejected: metric.WithAttributes( + attribute.String("error.type", string(JsonDecoderErrorTypeNullRejected)), + ), + JsonDecoderErrorTypeTrailingValue: metric.WithAttributes( + attribute.String("error.type", string(JsonDecoderErrorTypeTrailingValue)), + ), + JsonDecoderErrorTypeSyntaxError: metric.WithAttributes( + attribute.String("error.type", string(JsonDecoderErrorTypeSyntaxError)), + ), + JsonDecoderErrorTypeTypeError: metric.WithAttributes( + attribute.String("error.type", string(JsonDecoderErrorTypeTypeError)), + ), + JsonDecoderErrorTypeUnknownField: metric.WithAttributes( + attribute.String("error.type", string(JsonDecoderErrorTypeUnknownField)), + ), + JsonDecoderErrorTypeOther: metric.WithAttributes( + attribute.String("error.type", "_OTHER"), + ), + } + return nil +} + +// RecordGocciaJsonDecoderOperationDuration records the given value into the histogram metric. +func (m *JsonDecoder) RecordGocciaJsonDecoderOperationDuration( + ctx context.Context, + value float64, +) { + m.gocciaJsonDecoderOperationDuration.Record( + ctx, + value, + ) +} + +// RecordGocciaJsonDecoderOperationDurationWithAttributes records the given value +// and attributes into the histogram metric. +func (m *JsonDecoder) RecordGocciaJsonDecoderOperationDurationWithAttributes( + ctx context.Context, + value float64, + attributes metric.MeasurementOption, +) { + m.gocciaJsonDecoderOperationDuration.RecordWithAttributes( + ctx, + value, + attributes, + ) +} + +// RecordGocciaJsonDecoderOperationDurationWithErrorType records the given value +// and the error type into the histogram metric. +func (m *JsonDecoder) RecordGocciaJsonDecoderOperationDurationWithErrorType( + ctx context.Context, + value float64, + errorType JsonDecoderErrorType, +) { + m.gocciaJsonDecoderOperationDuration.RecordWithAttributes( + ctx, + value, + m.jsonDecoderErrorTypes[errorType], + ) +} + +// RecordGocciaJsonDecoderInputSize records the given value into the histogram metric. +func (m *JsonDecoder) RecordGocciaJsonDecoderInputSize( + ctx context.Context, + value int64, +) { + m.gocciaJsonDecoderInputSize.Record( + ctx, + value, + ) +} + +// RecordGocciaJsonDecoderInputSizeWithAttributes records the given value +// and attributes into the histogram metric. +func (m *JsonDecoder) RecordGocciaJsonDecoderInputSizeWithAttributes( + ctx context.Context, + value int64, + attributes metric.MeasurementOption, +) { + m.gocciaJsonDecoderInputSize.RecordWithAttributes( + ctx, + value, + attributes, + ) +} + +// RecordGocciaJsonDecoderInputSizeWithErrorType records the given value +// and the error type into the histogram metric. +func (m *JsonDecoder) RecordGocciaJsonDecoderInputSizeWithErrorType( + ctx context.Context, + value int64, + errorType JsonDecoderErrorType, +) { + m.gocciaJsonDecoderInputSize.RecordWithAttributes( + ctx, + value, + m.jsonDecoderErrorTypes[errorType], + ) +} diff --git a/processor/metrics/json_encoder.metrics.go b/processor/metrics/json_encoder.metrics.go new file mode 100644 index 0000000..23691d2 --- /dev/null +++ b/processor/metrics/json_encoder.metrics.go @@ -0,0 +1,131 @@ +// Code generated by metrics-gen. DO NOT EDIT. + +package metrics + +import ( + "context" + "github.com/FerroO2000/goccia/internal/telemetry" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/metric" +) + +// JsonEncoder structs contains the metrics for the json_encoder group. +type JsonEncoder struct { + gocciaJsonEncoderOperationDuration *telemetry.FloatHistogram + gocciaJsonEncoderOutputSize *telemetry.IntHistogram + jsonEncoderErrorTypes map[JsonEncoderErrorType]metric.MeasurementOption +} + +// NewJsonEncoder returns a new instance of the JsonEncoder struct. +func NewJsonEncoder() *JsonEncoder { + return &JsonEncoder{} +} + +// InitMetrics initializes the metrics for the JsonEncoder. +// It uses the given telemetry instance to create the metrics. +// +// It must be called before using the metrics. +func (m *JsonEncoder) InitMetrics(tel *telemetry.Telemetry) error { + var err error + + // Initialize goccia.json.encoder.operation.duration metric + m.gocciaJsonEncoderOperationDuration, err = tel.NewFloatHistogramMetric( + "goccia.json.encoder.operation.duration", + metric.WithUnit("s"), + metric.WithExplicitBucketBoundaries( + 1e-06, 2.5e-06, 5e-06, 1e-05, 2.5e-05, 5e-05, 0.0001, 0.00025, 0.0005, 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, + ), + ) + if err != nil { + return err + } + + // Initialize goccia.json.encoder.output.size metric + m.gocciaJsonEncoderOutputSize, err = tel.NewIntHistogramMetric( + "goccia.json.encoder.output.size", + metric.WithUnit("By"), + ) + if err != nil { + return err + } + + m.jsonEncoderErrorTypes = map[JsonEncoderErrorType]metric.MeasurementOption{ + JsonEncoderErrorTypeUnsupportedType: metric.WithAttributes( + attribute.String("error.type", string(JsonEncoderErrorTypeUnsupportedType)), + ), + JsonEncoderErrorTypeUnsupportedValue: metric.WithAttributes( + attribute.String("error.type", string(JsonEncoderErrorTypeUnsupportedValue)), + ), + JsonEncoderErrorTypeMarshalerError: metric.WithAttributes( + attribute.String("error.type", string(JsonEncoderErrorTypeMarshalerError)), + ), + JsonEncoderErrorTypeOther: metric.WithAttributes( + attribute.String("error.type", "_OTHER"), + ), + } + return nil +} + +// RecordGocciaJsonEncoderOperationDuration records the given value into the histogram metric. +func (m *JsonEncoder) RecordGocciaJsonEncoderOperationDuration( + ctx context.Context, + value float64, +) { + m.gocciaJsonEncoderOperationDuration.Record( + ctx, + value, + ) +} + +// RecordGocciaJsonEncoderOperationDurationWithAttributes records the given value +// and attributes into the histogram metric. +func (m *JsonEncoder) RecordGocciaJsonEncoderOperationDurationWithAttributes( + ctx context.Context, + value float64, + attributes metric.MeasurementOption, +) { + m.gocciaJsonEncoderOperationDuration.RecordWithAttributes( + ctx, + value, + attributes, + ) +} + +// RecordGocciaJsonEncoderOperationDurationWithErrorType records the given value +// and the error type into the histogram metric. +func (m *JsonEncoder) RecordGocciaJsonEncoderOperationDurationWithErrorType( + ctx context.Context, + value float64, + errorType JsonEncoderErrorType, +) { + m.gocciaJsonEncoderOperationDuration.RecordWithAttributes( + ctx, + value, + m.jsonEncoderErrorTypes[errorType], + ) +} + +// RecordGocciaJsonEncoderOutputSize records the given value into the histogram metric. +func (m *JsonEncoder) RecordGocciaJsonEncoderOutputSize( + ctx context.Context, + value int64, +) { + m.gocciaJsonEncoderOutputSize.Record( + ctx, + value, + ) +} + +// RecordGocciaJsonEncoderOutputSizeWithAttributes records the given value +// and attributes into the histogram metric. +func (m *JsonEncoder) RecordGocciaJsonEncoderOutputSizeWithAttributes( + ctx context.Context, + value int64, + attributes metric.MeasurementOption, +) { + m.gocciaJsonEncoderOutputSize.RecordWithAttributes( + ctx, + value, + attributes, + ) +} diff --git a/processor/metrics/rob_stage.doc.md b/processor/metrics/rob_stage.doc.md deleted file mode 100644 index e215875..0000000 --- a/processor/metrics/rob_stage.doc.md +++ /dev/null @@ -1,9 +0,0 @@ -| Name | Type | Data Type | Description | -|---------|---------|---------|---------| -| ordered_messages | `counter` | `integer` | - | -| primary_enqueued_messages | `counter` | `integer` | - | -| auxiliary_enqueued_messages | `counter` | `integer` | - | -| out_of_order_sequence_number | `counter` | `integer` | - | -| duplicated_sequence_number | `counter` | `integer` | - | -| invalid_sequence_number | `counter` | `integer` | - | -| resets | `counter` | `integer` | - | diff --git a/processor/metrics/router_stage.doc.md b/processor/metrics/router_stage.doc.md deleted file mode 100644 index a895ea5..0000000 --- a/processor/metrics/router_stage.doc.md +++ /dev/null @@ -1,3 +0,0 @@ -| Name | Type | Data Type | Description | -|---------|---------|---------|---------| -| total_routed_messages | `counter` | `integer` | - | diff --git a/processor/metrics/spec.yaml b/processor/metrics/spec.yaml index 0d8a80c..f11234a 100644 --- a/processor/metrics/spec.yaml +++ b/processor/metrics/spec.yaml @@ -1,4 +1,35 @@ package: metrics + +bucket_bounds_sets: + - name: micro_to_10_second + lower_bound: 0.000001 + upper_bound: 10 + +error_types: + - name: json_decoder + errors: + - name: input_too_large + value: goccia.json.input_too_large + - name: null_rejected + value: goccia.json.null_rejected + - name: trailing_value + value: goccia.json.trailing_value + - name: syntax_error + value: goccia.json.syntax_error + - name: type_error + value: goccia.json.type_error + - name: unknown_field + value: goccia.json.unknown_field + + - name: json_encoder + errors: + - name: unsupported_type + value: goccia.json.unsupported_type + - name: unsupported_value + value: goccia.json.unsupported_value + - name: marshaler_error + value: goccia.json.marshaler_error + groups: - name: aggregate_stage metrics: @@ -17,6 +48,33 @@ groups: - name: filtered_messages type: counter + - name: json_encoder + metrics: + - name: goccia.json.encoder.operation.duration + type: histogram + data_type: float + unit: s + bucket_bounds_set: micro_to_10_second + error_type: json_encoder + + - name: goccia.json.encoder.output.size + type: histogram + unit: By + + - name: json_decoder + metrics: + - name: goccia.json.decoder.operation.duration + type: histogram + data_type: float + unit: s + bucket_bounds_set: micro_to_10_second + error_type: json_decoder + + - name: goccia.json.decoder.input.size + type: histogram + unit: By + error_type: json_decoder + - name: tee_stage metrics: - name: cloned_messages diff --git a/processor/metrics/tee_stage.doc.md b/processor/metrics/tee_stage.doc.md deleted file mode 100644 index a93ecd9..0000000 --- a/processor/metrics/tee_stage.doc.md +++ /dev/null @@ -1,3 +0,0 @@ -| Name | Type | Data Type | Description | -|---------|---------|---------|---------| -| cloned_messages | `counter` | `integer` | - |