From 7cfaa6b213c31d503ee238bfa029ac87c5bf95ce Mon Sep 17 00:00:00 2001 From: Alan Shaw Date: Tue, 22 Sep 2026 07:00:05 +0100 Subject: [PATCH 1/3] feat: add command for retrieving metrics samples for a space --- commands/metrics/cbor_gen.go | 685 +++++++++++++++++++++++++++++++++ commands/metrics/codec_test.go | 104 +++++ commands/metrics/gen/main.go | 43 +++ commands/metrics/json_gen.go | 600 +++++++++++++++++++++++++++++ commands/metrics/sample.go | 36 ++ commands/metrics/types.go | 70 ++++ 6 files changed, 1538 insertions(+) create mode 100644 commands/metrics/cbor_gen.go create mode 100644 commands/metrics/codec_test.go create mode 100644 commands/metrics/gen/main.go create mode 100644 commands/metrics/json_gen.go create mode 100644 commands/metrics/sample.go create mode 100644 commands/metrics/types.go diff --git a/commands/metrics/cbor_gen.go b/commands/metrics/cbor_gen.go new file mode 100644 index 0000000..5247a22 --- /dev/null +++ b/commands/metrics/cbor_gen.go @@ -0,0 +1,685 @@ +//go:build !codegen + +// Code generated by github.com/whyrusleeping/cbor-gen. DO NOT EDIT. + +package metrics + +import ( + "fmt" + "io" + "math" + "sort" + + cid "github.com/ipfs/go-cid" + cbg "github.com/whyrusleeping/cbor-gen" + xerrors "golang.org/x/xerrors" +) + +var _ = xerrors.Errorf +var _ = cid.Undef +var _ = math.E +var _ = sort.Sort + +func (t *SampleArguments) MarshalCBOR(w io.Writer) error { + if t == nil { + _, err := w.Write(cbg.CborNull) + return err + } + + cw := cbg.NewCborWriter(w) + + if _, err := cw.Write([]byte{163}); err != nil { + return err + } + + // t.To (int64) (int64) + if len("to") > 8192 { + return xerrors.Errorf("Value in field \"to\" was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("to"))); err != nil { + return err + } + if _, err := cw.WriteString(string("to")); err != nil { + return err + } + + if t.To >= 0 { + if err := cw.WriteMajorTypeHeader(cbg.MajUnsignedInt, uint64(t.To)); err != nil { + return err + } + } else { + if err := cw.WriteMajorTypeHeader(cbg.MajNegativeInt, uint64(-t.To-1)); err != nil { + return err + } + } + + // t.From (int64) (int64) + if len("from") > 8192 { + return xerrors.Errorf("Value in field \"from\" was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("from"))); err != nil { + return err + } + if _, err := cw.WriteString(string("from")); err != nil { + return err + } + + if t.From >= 0 { + if err := cw.WriteMajorTypeHeader(cbg.MajUnsignedInt, uint64(t.From)); err != nil { + return err + } + } else { + if err := cw.WriteMajorTypeHeader(cbg.MajNegativeInt, uint64(-t.From-1)); err != nil { + return err + } + } + + // t.Window (int64) (int64) + if len("window") > 8192 { + return xerrors.Errorf("Value in field \"window\" was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("window"))); err != nil { + return err + } + if _, err := cw.WriteString(string("window")); err != nil { + return err + } + + if t.Window >= 0 { + if err := cw.WriteMajorTypeHeader(cbg.MajUnsignedInt, uint64(t.Window)); err != nil { + return err + } + } else { + if err := cw.WriteMajorTypeHeader(cbg.MajNegativeInt, uint64(-t.Window-1)); err != nil { + return err + } + } + + return nil +} + +func (t *SampleArguments) UnmarshalCBOR(r io.Reader) (err error) { + *t = SampleArguments{} + + cr := cbg.NewCborReader(r) + + maj, extra, err := cr.ReadHeader() + if err != nil { + return err + } + defer func() { + if err == io.EOF { + err = io.ErrUnexpectedEOF + } + }() + + if maj != cbg.MajMap { + return fmt.Errorf("cbor input should be of type map") + } + + if extra > cbg.MaxLength { + return fmt.Errorf("SampleArguments: map struct too large (%d)", extra) + } + + n := extra + + nameBuf := make([]byte, 6) + for i := uint64(0); i < n; i++ { + nameLen, ok, err := cbg.ReadFullStringIntoBuf(cr, nameBuf, 8192) + if err != nil { + return err + } + + if !ok { + // Field doesn't exist on this type, so ignore it + if err := cbg.ScanForLinks(cr, func(cid.Cid) {}); err != nil { + return err + } + continue + } + + switch string(nameBuf[:nameLen]) { + // t.To (int64) (int64) + case "to": + { + maj, extra, err := cr.ReadHeader() + if err != nil { + return err + } + var extraI int64 + switch maj { + case cbg.MajUnsignedInt: + extraI = int64(extra) + if extraI < 0 { + return fmt.Errorf("int64 positive overflow") + } + case cbg.MajNegativeInt: + extraI = int64(extra) + if extraI < 0 { + return fmt.Errorf("int64 negative overflow") + } + extraI = -1 - extraI + default: + return fmt.Errorf("wrong type for int64 field: %d", maj) + } + + t.To = int64(extraI) + } + // t.From (int64) (int64) + case "from": + { + maj, extra, err := cr.ReadHeader() + if err != nil { + return err + } + var extraI int64 + switch maj { + case cbg.MajUnsignedInt: + extraI = int64(extra) + if extraI < 0 { + return fmt.Errorf("int64 positive overflow") + } + case cbg.MajNegativeInt: + extraI = int64(extra) + if extraI < 0 { + return fmt.Errorf("int64 negative overflow") + } + extraI = -1 - extraI + default: + return fmt.Errorf("wrong type for int64 field: %d", maj) + } + + t.From = int64(extraI) + } + // t.Window (int64) (int64) + case "window": + { + maj, extra, err := cr.ReadHeader() + if err != nil { + return err + } + var extraI int64 + switch maj { + case cbg.MajUnsignedInt: + extraI = int64(extra) + if extraI < 0 { + return fmt.Errorf("int64 positive overflow") + } + case cbg.MajNegativeInt: + extraI = int64(extra) + if extraI < 0 { + return fmt.Errorf("int64 negative overflow") + } + extraI = -1 - extraI + default: + return fmt.Errorf("wrong type for int64 field: %d", maj) + } + + t.Window = int64(extraI) + } + + default: + // Field doesn't exist on this type, so ignore it + if err := cbg.ScanForLinks(r, func(cid.Cid) {}); err != nil { + return err + } + } + } + + return nil +} +func (t *SampleOK) MarshalCBOR(w io.Writer) error { + if t == nil { + _, err := w.Write(cbg.CborNull) + return err + } + + cw := cbg.NewCborWriter(w) + + if _, err := cw.Write([]byte{164}); err != nil { + return err + } + + // t.To (int64) (int64) + if len("to") > 8192 { + return xerrors.Errorf("Value in field \"to\" was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("to"))); err != nil { + return err + } + if _, err := cw.WriteString(string("to")); err != nil { + return err + } + + if t.To >= 0 { + if err := cw.WriteMajorTypeHeader(cbg.MajUnsignedInt, uint64(t.To)); err != nil { + return err + } + } else { + if err := cw.WriteMajorTypeHeader(cbg.MajNegativeInt, uint64(-t.To-1)); err != nil { + return err + } + } + + // t.From (int64) (int64) + if len("from") > 8192 { + return xerrors.Errorf("Value in field \"from\" was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("from"))); err != nil { + return err + } + if _, err := cw.WriteString(string("from")); err != nil { + return err + } + + if t.From >= 0 { + if err := cw.WriteMajorTypeHeader(cbg.MajUnsignedInt, uint64(t.From)); err != nil { + return err + } + } else { + if err := cw.WriteMajorTypeHeader(cbg.MajNegativeInt, uint64(-t.From-1)); err != nil { + return err + } + } + + // t.Window (int64) (int64) + if len("window") > 8192 { + return xerrors.Errorf("Value in field \"window\" was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("window"))); err != nil { + return err + } + if _, err := cw.WriteString(string("window")); err != nil { + return err + } + + if t.Window >= 0 { + if err := cw.WriteMajorTypeHeader(cbg.MajUnsignedInt, uint64(t.Window)); err != nil { + return err + } + } else { + if err := cw.WriteMajorTypeHeader(cbg.MajNegativeInt, uint64(-t.Window-1)); err != nil { + return err + } + } + + // t.Samples ([]metrics.SampleItem) (slice) + if len("samples") > 8192 { + return xerrors.Errorf("Value in field \"samples\" was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("samples"))); err != nil { + return err + } + if _, err := cw.WriteString(string("samples")); err != nil { + return err + } + + if len(t.Samples) > 8192 { + return xerrors.Errorf("Slice value in field t.Samples was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajArray, uint64(len(t.Samples))); err != nil { + return err + } + for _, v := range t.Samples { + if err := v.MarshalCBOR(cw); err != nil { + return err + } + + } + return nil +} + +func (t *SampleOK) UnmarshalCBOR(r io.Reader) (err error) { + *t = SampleOK{} + + cr := cbg.NewCborReader(r) + + maj, extra, err := cr.ReadHeader() + if err != nil { + return err + } + defer func() { + if err == io.EOF { + err = io.ErrUnexpectedEOF + } + }() + + if maj != cbg.MajMap { + return fmt.Errorf("cbor input should be of type map") + } + + if extra > cbg.MaxLength { + return fmt.Errorf("SampleOK: map struct too large (%d)", extra) + } + + n := extra + + nameBuf := make([]byte, 7) + for i := uint64(0); i < n; i++ { + nameLen, ok, err := cbg.ReadFullStringIntoBuf(cr, nameBuf, 8192) + if err != nil { + return err + } + + if !ok { + // Field doesn't exist on this type, so ignore it + if err := cbg.ScanForLinks(cr, func(cid.Cid) {}); err != nil { + return err + } + continue + } + + switch string(nameBuf[:nameLen]) { + // t.To (int64) (int64) + case "to": + { + maj, extra, err := cr.ReadHeader() + if err != nil { + return err + } + var extraI int64 + switch maj { + case cbg.MajUnsignedInt: + extraI = int64(extra) + if extraI < 0 { + return fmt.Errorf("int64 positive overflow") + } + case cbg.MajNegativeInt: + extraI = int64(extra) + if extraI < 0 { + return fmt.Errorf("int64 negative overflow") + } + extraI = -1 - extraI + default: + return fmt.Errorf("wrong type for int64 field: %d", maj) + } + + t.To = int64(extraI) + } + // t.From (int64) (int64) + case "from": + { + maj, extra, err := cr.ReadHeader() + if err != nil { + return err + } + var extraI int64 + switch maj { + case cbg.MajUnsignedInt: + extraI = int64(extra) + if extraI < 0 { + return fmt.Errorf("int64 positive overflow") + } + case cbg.MajNegativeInt: + extraI = int64(extra) + if extraI < 0 { + return fmt.Errorf("int64 negative overflow") + } + extraI = -1 - extraI + default: + return fmt.Errorf("wrong type for int64 field: %d", maj) + } + + t.From = int64(extraI) + } + // t.Window (int64) (int64) + case "window": + { + maj, extra, err := cr.ReadHeader() + if err != nil { + return err + } + var extraI int64 + switch maj { + case cbg.MajUnsignedInt: + extraI = int64(extra) + if extraI < 0 { + return fmt.Errorf("int64 positive overflow") + } + case cbg.MajNegativeInt: + extraI = int64(extra) + if extraI < 0 { + return fmt.Errorf("int64 negative overflow") + } + extraI = -1 - extraI + default: + return fmt.Errorf("wrong type for int64 field: %d", maj) + } + + t.Window = int64(extraI) + } + // t.Samples ([]metrics.SampleItem) (slice) + case "samples": + + maj, extra, err = cr.ReadHeader() + if err != nil { + return err + } + + if extra > 8192 { + return fmt.Errorf("t.Samples: array too large (%d)", extra) + } + + if maj != cbg.MajArray { + return fmt.Errorf("expected cbor array") + } + + if extra > 0 { + t.Samples = make([]SampleItem, extra) + } + + for i := 0; i < int(extra); i++ { + { + var maj byte + var extra uint64 + var err error + _ = maj + _ = extra + _ = err + + { + + if err := t.Samples[i].UnmarshalCBOR(cr); err != nil { + return xerrors.Errorf("unmarshaling t.Samples[i]: %w", err) + } + + } + + } + } + + default: + // Field doesn't exist on this type, so ignore it + if err := cbg.ScanForLinks(r, func(cid.Cid) {}); err != nil { + return err + } + } + } + + return nil +} +func (t *SampleItem) MarshalCBOR(w io.Writer) error { + if t == nil { + _, err := w.Write(cbg.CborNull) + return err + } + + cw := cbg.NewCborWriter(w) + + if _, err := cw.Write([]byte{163}); err != nil { + return err + } + + // t.Timestamp (int64) (int64) + if len("timestamp") > 8192 { + return xerrors.Errorf("Value in field \"timestamp\" was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("timestamp"))); err != nil { + return err + } + if _, err := cw.WriteString(string("timestamp")); err != nil { + return err + } + + if t.Timestamp >= 0 { + if err := cw.WriteMajorTypeHeader(cbg.MajUnsignedInt, uint64(t.Timestamp)); err != nil { + return err + } + } else { + if err := cw.WriteMajorTypeHeader(cbg.MajNegativeInt, uint64(-t.Timestamp-1)); err != nil { + return err + } + } + + // t.BytesStored (uint64) (uint64) + if len("bytesStored") > 8192 { + return xerrors.Errorf("Value in field \"bytesStored\" was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("bytesStored"))); err != nil { + return err + } + if _, err := cw.WriteString(string("bytesStored")); err != nil { + return err + } + + if err := cw.WriteMajorTypeHeader(cbg.MajUnsignedInt, uint64(t.BytesStored)); err != nil { + return err + } + + // t.BytesIngested (uint64) (uint64) + if len("bytesIngested") > 8192 { + return xerrors.Errorf("Value in field \"bytesIngested\" was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("bytesIngested"))); err != nil { + return err + } + if _, err := cw.WriteString(string("bytesIngested")); err != nil { + return err + } + + if err := cw.WriteMajorTypeHeader(cbg.MajUnsignedInt, uint64(t.BytesIngested)); err != nil { + return err + } + + return nil +} + +func (t *SampleItem) UnmarshalCBOR(r io.Reader) (err error) { + *t = SampleItem{} + + cr := cbg.NewCborReader(r) + + maj, extra, err := cr.ReadHeader() + if err != nil { + return err + } + defer func() { + if err == io.EOF { + err = io.ErrUnexpectedEOF + } + }() + + if maj != cbg.MajMap { + return fmt.Errorf("cbor input should be of type map") + } + + if extra > cbg.MaxLength { + return fmt.Errorf("SampleItem: map struct too large (%d)", extra) + } + + n := extra + + nameBuf := make([]byte, 13) + for i := uint64(0); i < n; i++ { + nameLen, ok, err := cbg.ReadFullStringIntoBuf(cr, nameBuf, 8192) + if err != nil { + return err + } + + if !ok { + // Field doesn't exist on this type, so ignore it + if err := cbg.ScanForLinks(cr, func(cid.Cid) {}); err != nil { + return err + } + continue + } + + switch string(nameBuf[:nameLen]) { + // t.Timestamp (int64) (int64) + case "timestamp": + { + maj, extra, err := cr.ReadHeader() + if err != nil { + return err + } + var extraI int64 + switch maj { + case cbg.MajUnsignedInt: + extraI = int64(extra) + if extraI < 0 { + return fmt.Errorf("int64 positive overflow") + } + case cbg.MajNegativeInt: + extraI = int64(extra) + if extraI < 0 { + return fmt.Errorf("int64 negative overflow") + } + extraI = -1 - extraI + default: + return fmt.Errorf("wrong type for int64 field: %d", maj) + } + + t.Timestamp = int64(extraI) + } + // t.BytesStored (uint64) (uint64) + case "bytesStored": + + { + + maj, extra, err = cr.ReadHeader() + if err != nil { + return err + } + if maj != cbg.MajUnsignedInt { + return fmt.Errorf("wrong type for uint64 field") + } + t.BytesStored = uint64(extra) + + } + // t.BytesIngested (uint64) (uint64) + case "bytesIngested": + + { + + maj, extra, err = cr.ReadHeader() + if err != nil { + return err + } + if maj != cbg.MajUnsignedInt { + return fmt.Errorf("wrong type for uint64 field") + } + t.BytesIngested = uint64(extra) + + } + + default: + // Field doesn't exist on this type, so ignore it + if err := cbg.ScanForLinks(r, func(cid.Cid) {}); err != nil { + return err + } + } + } + + return nil +} diff --git a/commands/metrics/codec_test.go b/commands/metrics/codec_test.go new file mode 100644 index 0000000..fc95590 --- /dev/null +++ b/commands/metrics/codec_test.go @@ -0,0 +1,104 @@ +//go:build !codegen + +package metrics_test + +import ( + "bytes" + "io" + "reflect" + "testing" + + "github.com/fil-forge/libforge/commands/metrics" + "github.com/stretchr/testify/require" +) + +type wire interface { + MarshalCBOR(w io.Writer) error + UnmarshalCBOR(r io.Reader) error + MarshalDagJSON(w io.Writer) error + UnmarshalDagJSON(r io.Reader) error +} + +// roundTrip encodes in as CBOR and DAG-JSON, decodes each and asserts equality. +// It returns the DAG-JSON encoding. +func roundTrip[T any, PT interface { + *T + wire +}](t *testing.T, in PT) string { + t.Helper() + var cb bytes.Buffer + require.NoError(t, in.MarshalCBOR(&cb)) + outCBOR := PT(new(T)) + require.NoError(t, outCBOR.UnmarshalCBOR(bytes.NewReader(cb.Bytes()))) + require.True(t, reflect.DeepEqual(*in, *outCBOR), "CBOR round-trip mismatch:\n got %#v\nwant %#v", *outCBOR, *in) + + var jb bytes.Buffer + require.NoError(t, in.MarshalDagJSON(&jb)) + outJSON := PT(new(T)) + require.NoError(t, outJSON.UnmarshalDagJSON(bytes.NewReader(jb.Bytes())), "json: %s", jb.String()) + require.True(t, reflect.DeepEqual(*in, *outJSON), "DAG-JSON round-trip mismatch:\n got %#v\nwant %#v", *outJSON, *in) + return jb.String() +} + +func TestSampleArgumentsRoundTrip(t *testing.T) { + in := &metrics.SampleArguments{From: 1700000000, To: 1700003600, Window: 3600} + require.Equal(t, `{"from":1700000000,"to":1700003600,"window":3600}`, roundTrip(t, in)) +} + +func TestSampleOKRoundTrip(t *testing.T) { + in := &metrics.SampleOK{ + From: 1700000000, + To: 1700007200, + Window: 3600, + Samples: []metrics.SampleItem{ + {Timestamp: 1700003600, BytesStored: 1024, BytesIngested: 1024}, + {Timestamp: 1700007200, BytesStored: 512, BytesIngested: 0}, + }, + } + require.Equal(t, + `{"from":1700000000,"samples":[`+ + `{"bytesIngested":1024,"bytesStored":1024,"timestamp":1700003600},`+ + `{"bytesIngested":0,"bytesStored":512,"timestamp":1700007200}`+ + `],"to":1700007200,"window":3600}`, + roundTrip(t, in)) +} + +// A range entirely in the future carries no samples, so the empty series has to +// survive both codecs. +func TestSampleOKEmptySeriesRoundTrip(t *testing.T) { + in := &metrics.SampleOK{From: 1700000000, To: 1700000000, Window: 3600} + require.Equal(t, `{"from":1700000000,"samples":[],"to":1700000000,"window":3600}`, roundTrip(t, in)) + + var cb bytes.Buffer + require.NoError(t, in.MarshalCBOR(&cb)) + var out metrics.SampleOK + require.NoError(t, out.UnmarshalCBOR(bytes.NewReader(cb.Bytes()))) + require.Empty(t, out.Samples) +} + +// 768 samples is a 32 day range at hourly granularity, the largest series +// consumers ask for. Both codecs cap arrays at 8192 elements, so this stays +// well inside the limit — but the encoded size is worth knowing. +func TestSampleOKLargeSeriesRoundTrip(t *testing.T) { + const ( + hour = int64(3600) + count = 768 + ) + in := &metrics.SampleOK{From: 1700000000, Window: hour} + for i := range count { + in.Samples = append(in.Samples, metrics.SampleItem{ + Timestamp: in.From + int64(i+1)*hour, + BytesStored: uint64(i) * 1 << 20, + BytesIngested: 1 << 20, + }) + } + in.To = in.From + count*hour + + var cb bytes.Buffer + require.NoError(t, in.MarshalCBOR(&cb)) + var out metrics.SampleOK + require.NoError(t, out.UnmarshalCBOR(bytes.NewReader(cb.Bytes()))) + require.Len(t, out.Samples, count) + require.Equal(t, *in, out) + t.Logf("%d samples encode to %d bytes of CBOR", count, cb.Len()) +} diff --git a/commands/metrics/gen/main.go b/commands/metrics/gen/main.go new file mode 100644 index 0000000..c871b91 --- /dev/null +++ b/commands/metrics/gen/main.go @@ -0,0 +1,43 @@ +//go:generate go run -tags codegen . + +package main + +import ( + "os" + + jsg "github.com/alanshaw/dag-json-gen" + "github.com/fil-forge/libforge/commands/metrics" + cbg "github.com/whyrusleeping/cbor-gen" +) + +const buildTag = "//go:build !codegen\n\n" + +func tag(path string) { + data, err := os.ReadFile(path) + if err != nil { + panic(err) + } + if err := os.WriteFile(path, append([]byte(buildTag), data...), 0644); err != nil { + panic(err) + } +} + +func main() { + models := []any{ + metrics.SampleArguments{}, + metrics.SampleOK{}, + metrics.SampleItem{}, + } + const ( + cborFile = "../cbor_gen.go" + jsonFile = "../json_gen.go" + ) + if err := cbg.WriteMapEncodersToFile(cborFile, "metrics", models...); err != nil { + panic(err) + } + if err := jsg.WriteMapEncodersToFile(jsonFile, "metrics", models...); err != nil { + panic(err) + } + tag(cborFile) + tag(jsonFile) +} diff --git a/commands/metrics/json_gen.go b/commands/metrics/json_gen.go new file mode 100644 index 0000000..264f70e --- /dev/null +++ b/commands/metrics/json_gen.go @@ -0,0 +1,600 @@ +//go:build !codegen + +// Code generated by github.com/alanshaw/dag-json-gen. DO NOT EDIT. + +package metrics + +import ( + "errors" + "fmt" + "io" + "math" + "sort" + + jsg "github.com/alanshaw/dag-json-gen" + cid "github.com/ipfs/go-cid" +) + +var _ = cid.Undef +var _ = math.E +var _ = sort.Sort +var _ = errors.Is + +func (t *SampleArguments) MarshalDagJSON(w io.Writer) error { + jw := jsg.NewDagJsonWriter(w) + if t == nil { + err := jw.WriteNull() + return err + } + if err := jw.WriteObjectOpen(); err != nil { + return err + } + written := false + + // t.From (int64) (int64) + if len("from") > 8192 { + return fmt.Errorf("string in field \"from\" was too long") + } + if err := jw.WriteString(string("from")); err != nil { + return fmt.Errorf("writing string for field \"from\": %w", err) + } + if err := jw.WriteObjectColon(); err != nil { + return err + } + + if err := jw.WriteInt64(int64(t.From)); err != nil { + return fmt.Errorf("writing int64 for field t.From: %w", err) + } + + written = true + if written { + if err := jw.WriteComma(); err != nil { + return err + } + } + + // t.To (int64) (int64) + if len("to") > 8192 { + return fmt.Errorf("string in field \"to\" was too long") + } + if err := jw.WriteString(string("to")); err != nil { + return fmt.Errorf("writing string for field \"to\": %w", err) + } + if err := jw.WriteObjectColon(); err != nil { + return err + } + + if err := jw.WriteInt64(int64(t.To)); err != nil { + return fmt.Errorf("writing int64 for field t.To: %w", err) + } + + written = true + if written { + if err := jw.WriteComma(); err != nil { + return err + } + } + + // t.Window (int64) (int64) + if len("window") > 8192 { + return fmt.Errorf("string in field \"window\" was too long") + } + if err := jw.WriteString(string("window")); err != nil { + return fmt.Errorf("writing string for field \"window\": %w", err) + } + if err := jw.WriteObjectColon(); err != nil { + return err + } + + if err := jw.WriteInt64(int64(t.Window)); err != nil { + return fmt.Errorf("writing int64 for field t.Window: %w", err) + } + + if err := jw.WriteObjectClose(); err != nil { + return err + } + return nil +} +func (t *SampleArguments) UnmarshalDagJSON(r io.Reader) (err error) { + *t = SampleArguments{} + + jr := jsg.NewDagJsonReader(r) + defer func() { + if err == io.EOF { + err = io.ErrUnexpectedEOF + } + }() + if err := jr.ReadObjectOpen(); err != nil { + return fmt.Errorf("reading object open for SampleArguments: %w", err) + } + close, err := jr.PeekObjectClose() + if err != nil { + return fmt.Errorf("peeking object close for SampleArguments: %w", err) + } + if close { + if err := jr.ReadObjectClose(); err != nil { + return fmt.Errorf("reading object close for SampleArguments: %w", err) + } + } else { + for i := uint64(0); i < 8192; i++ { + name, err := jr.ReadString(8192) + if err != nil { + if errors.Is(err, jsg.ErrLimitExceeded) { + return fmt.Errorf("reading string for field SampleArguments: string too large") + } + return fmt.Errorf("reading string for field SampleArguments: %w", err) + } + if err := jr.ReadObjectColon(); err != nil { + return fmt.Errorf("reading object colon for field SampleArguments: %w", err) + } + switch name { + + // t.From (int64) (int64) + case "from": + { + + nval, err := jr.ReadNumberAsInt64() + if err != nil { + return fmt.Errorf("reading int64 for field t.From: %w", err) + } + t.From = int64(nval) + + } + + // t.To (int64) (int64) + case "to": + { + + nval, err := jr.ReadNumberAsInt64() + if err != nil { + return fmt.Errorf("reading int64 for field t.To: %w", err) + } + t.To = int64(nval) + + } + + // t.Window (int64) (int64) + case "window": + { + + nval, err := jr.ReadNumberAsInt64() + if err != nil { + return fmt.Errorf("reading int64 for field t.Window: %w", err) + } + t.Window = int64(nval) + + } + default: + // Field doesn't exist on this type, so ignore it + if err := jr.DiscardType(); err != nil { + return fmt.Errorf("ignoring field %s for SampleArguments: %w", name, err) + } + } + + close, err := jr.ReadObjectCloseOrComma() + if err != nil { + return fmt.Errorf("reading object close or comma for field SampleArguments: %w", err) + } + if close { + break + } + if i == 8192-1 { + return fmt.Errorf("map too large for SampleArguments") + } + } + } + + return nil +} +func (t *SampleOK) MarshalDagJSON(w io.Writer) error { + jw := jsg.NewDagJsonWriter(w) + if t == nil { + err := jw.WriteNull() + return err + } + if err := jw.WriteObjectOpen(); err != nil { + return err + } + written := false + + // t.From (int64) (int64) + if len("from") > 8192 { + return fmt.Errorf("string in field \"from\" was too long") + } + if err := jw.WriteString(string("from")); err != nil { + return fmt.Errorf("writing string for field \"from\": %w", err) + } + if err := jw.WriteObjectColon(); err != nil { + return err + } + + if err := jw.WriteInt64(int64(t.From)); err != nil { + return fmt.Errorf("writing int64 for field t.From: %w", err) + } + + written = true + if written { + if err := jw.WriteComma(); err != nil { + return err + } + } + + // t.Samples ([]metrics.SampleItem) (slice) + if len("samples") > 8192 { + return fmt.Errorf("string in field \"samples\" was too long") + } + if err := jw.WriteString(string("samples")); err != nil { + return fmt.Errorf("writing string for field \"samples\": %w", err) + } + if err := jw.WriteObjectColon(); err != nil { + return err + } + if len(t.Samples) > 8192 { + return fmt.Errorf("slice value in field t.Samples was too long") + } + + if err := jw.WriteArrayOpen(); err != nil { + return fmt.Errorf("writing array open for field t.Samples: %w", err) + } + for i, v := range t.Samples { + if i > 0 { + if err := jw.WriteComma(); err != nil { + return fmt.Errorf("writing comma for field t.Samples: %w", err) + } + } + if err := v.MarshalDagJSON(jw); err != nil { + return fmt.Errorf("marshaling field v: %w", err) + } + } + if err := jw.WriteArrayClose(); err != nil { + return fmt.Errorf("writing array close for field t.Samples: %w", err) + } + + written = true + if written { + if err := jw.WriteComma(); err != nil { + return err + } + } + + // t.To (int64) (int64) + if len("to") > 8192 { + return fmt.Errorf("string in field \"to\" was too long") + } + if err := jw.WriteString(string("to")); err != nil { + return fmt.Errorf("writing string for field \"to\": %w", err) + } + if err := jw.WriteObjectColon(); err != nil { + return err + } + + if err := jw.WriteInt64(int64(t.To)); err != nil { + return fmt.Errorf("writing int64 for field t.To: %w", err) + } + + written = true + if written { + if err := jw.WriteComma(); err != nil { + return err + } + } + + // t.Window (int64) (int64) + if len("window") > 8192 { + return fmt.Errorf("string in field \"window\" was too long") + } + if err := jw.WriteString(string("window")); err != nil { + return fmt.Errorf("writing string for field \"window\": %w", err) + } + if err := jw.WriteObjectColon(); err != nil { + return err + } + + if err := jw.WriteInt64(int64(t.Window)); err != nil { + return fmt.Errorf("writing int64 for field t.Window: %w", err) + } + + if err := jw.WriteObjectClose(); err != nil { + return err + } + return nil +} +func (t *SampleOK) UnmarshalDagJSON(r io.Reader) (err error) { + *t = SampleOK{} + + jr := jsg.NewDagJsonReader(r) + defer func() { + if err == io.EOF { + err = io.ErrUnexpectedEOF + } + }() + if err := jr.ReadObjectOpen(); err != nil { + return fmt.Errorf("reading object open for SampleOK: %w", err) + } + close, err := jr.PeekObjectClose() + if err != nil { + return fmt.Errorf("peeking object close for SampleOK: %w", err) + } + if close { + if err := jr.ReadObjectClose(); err != nil { + return fmt.Errorf("reading object close for SampleOK: %w", err) + } + } else { + for i := uint64(0); i < 8192; i++ { + name, err := jr.ReadString(8192) + if err != nil { + if errors.Is(err, jsg.ErrLimitExceeded) { + return fmt.Errorf("reading string for field SampleOK: string too large") + } + return fmt.Errorf("reading string for field SampleOK: %w", err) + } + if err := jr.ReadObjectColon(); err != nil { + return fmt.Errorf("reading object colon for field SampleOK: %w", err) + } + switch name { + + // t.From (int64) (int64) + case "from": + { + + nval, err := jr.ReadNumberAsInt64() + if err != nil { + return fmt.Errorf("reading int64 for field t.From: %w", err) + } + t.From = int64(nval) + + } + + // t.Samples ([]metrics.SampleItem) (slice) + case "samples": + { + + if err := jr.ReadArrayOpen(); err != nil { + return fmt.Errorf("reading array open for field t.Samples: %w", err) + } + + close, err := jr.PeekArrayClose() + if err != nil { + return fmt.Errorf("peeking array close for field t.Samples: %w", err) + } + if close { + if err := jr.ReadArrayClose(); err != nil { + return fmt.Errorf("reading array close for field t.Samples: %w", err) + } + + } else { + for i := 0; i < 8192; i++ { + item := make([]SampleItem, 1) + + if err := item[0].UnmarshalDagJSON(jr); err != nil { + return fmt.Errorf("unmarshaling item[0]: %w", err) + } + + t.Samples = append(t.Samples, item[0]) + + close, err := jr.ReadArrayCloseOrComma() + if err != nil { + return fmt.Errorf("reading array close or comma for field t.Samples: %w", err) + } + if close { + break + } + if i == 8192-1 { + return fmt.Errorf("reading array for field t.Samples: slice too large") + } + } + } + + } + + // t.To (int64) (int64) + case "to": + { + + nval, err := jr.ReadNumberAsInt64() + if err != nil { + return fmt.Errorf("reading int64 for field t.To: %w", err) + } + t.To = int64(nval) + + } + + // t.Window (int64) (int64) + case "window": + { + + nval, err := jr.ReadNumberAsInt64() + if err != nil { + return fmt.Errorf("reading int64 for field t.Window: %w", err) + } + t.Window = int64(nval) + + } + default: + // Field doesn't exist on this type, so ignore it + if err := jr.DiscardType(); err != nil { + return fmt.Errorf("ignoring field %s for SampleOK: %w", name, err) + } + } + + close, err := jr.ReadObjectCloseOrComma() + if err != nil { + return fmt.Errorf("reading object close or comma for field SampleOK: %w", err) + } + if close { + break + } + if i == 8192-1 { + return fmt.Errorf("map too large for SampleOK") + } + } + } + + return nil +} +func (t *SampleItem) MarshalDagJSON(w io.Writer) error { + jw := jsg.NewDagJsonWriter(w) + if t == nil { + err := jw.WriteNull() + return err + } + if err := jw.WriteObjectOpen(); err != nil { + return err + } + written := false + + // t.BytesIngested (uint64) (uint64) + if len("bytesIngested") > 8192 { + return fmt.Errorf("string in field \"bytesIngested\" was too long") + } + if err := jw.WriteString(string("bytesIngested")); err != nil { + return fmt.Errorf("writing string for field \"bytesIngested\": %w", err) + } + if err := jw.WriteObjectColon(); err != nil { + return err + } + + if err := jw.WriteUint64(uint64(t.BytesIngested)); err != nil { + return fmt.Errorf("writing uint64 for field t.BytesIngested: %w", err) + } + + written = true + if written { + if err := jw.WriteComma(); err != nil { + return err + } + } + + // t.BytesStored (uint64) (uint64) + if len("bytesStored") > 8192 { + return fmt.Errorf("string in field \"bytesStored\" was too long") + } + if err := jw.WriteString(string("bytesStored")); err != nil { + return fmt.Errorf("writing string for field \"bytesStored\": %w", err) + } + if err := jw.WriteObjectColon(); err != nil { + return err + } + + if err := jw.WriteUint64(uint64(t.BytesStored)); err != nil { + return fmt.Errorf("writing uint64 for field t.BytesStored: %w", err) + } + + written = true + if written { + if err := jw.WriteComma(); err != nil { + return err + } + } + + // t.Timestamp (int64) (int64) + if len("timestamp") > 8192 { + return fmt.Errorf("string in field \"timestamp\" was too long") + } + if err := jw.WriteString(string("timestamp")); err != nil { + return fmt.Errorf("writing string for field \"timestamp\": %w", err) + } + if err := jw.WriteObjectColon(); err != nil { + return err + } + + if err := jw.WriteInt64(int64(t.Timestamp)); err != nil { + return fmt.Errorf("writing int64 for field t.Timestamp: %w", err) + } + + if err := jw.WriteObjectClose(); err != nil { + return err + } + return nil +} +func (t *SampleItem) UnmarshalDagJSON(r io.Reader) (err error) { + *t = SampleItem{} + + jr := jsg.NewDagJsonReader(r) + defer func() { + if err == io.EOF { + err = io.ErrUnexpectedEOF + } + }() + if err := jr.ReadObjectOpen(); err != nil { + return fmt.Errorf("reading object open for SampleItem: %w", err) + } + close, err := jr.PeekObjectClose() + if err != nil { + return fmt.Errorf("peeking object close for SampleItem: %w", err) + } + if close { + if err := jr.ReadObjectClose(); err != nil { + return fmt.Errorf("reading object close for SampleItem: %w", err) + } + } else { + for i := uint64(0); i < 8192; i++ { + name, err := jr.ReadString(8192) + if err != nil { + if errors.Is(err, jsg.ErrLimitExceeded) { + return fmt.Errorf("reading string for field SampleItem: string too large") + } + return fmt.Errorf("reading string for field SampleItem: %w", err) + } + if err := jr.ReadObjectColon(); err != nil { + return fmt.Errorf("reading object colon for field SampleItem: %w", err) + } + switch name { + + // t.BytesIngested (uint64) (uint64) + case "bytesIngested": + { + + nval, err := jr.ReadNumberAsUint64() + if err != nil { + return fmt.Errorf("reading uint64 for field t.BytesIngested: %w", err) + } + t.BytesIngested = uint64(nval) + + } + + // t.BytesStored (uint64) (uint64) + case "bytesStored": + { + + nval, err := jr.ReadNumberAsUint64() + if err != nil { + return fmt.Errorf("reading uint64 for field t.BytesStored: %w", err) + } + t.BytesStored = uint64(nval) + + } + + // t.Timestamp (int64) (int64) + case "timestamp": + { + + nval, err := jr.ReadNumberAsInt64() + if err != nil { + return fmt.Errorf("reading int64 for field t.Timestamp: %w", err) + } + t.Timestamp = int64(nval) + + } + default: + // Field doesn't exist on this type, so ignore it + if err := jr.DiscardType(); err != nil { + return fmt.Errorf("ignoring field %s for SampleItem: %w", name, err) + } + } + + close, err := jr.ReadObjectCloseOrComma() + if err != nil { + return fmt.Errorf("reading object close or comma for field SampleItem: %w", err) + } + if close { + break + } + if i == 8192-1 { + return fmt.Errorf("map too large for SampleItem") + } + } + } + + return nil +} diff --git a/commands/metrics/sample.go b/commands/metrics/sample.go new file mode 100644 index 0000000..4b7e40e --- /dev/null +++ b/commands/metrics/sample.go @@ -0,0 +1,36 @@ +//go:build !codegen + +package metrics + +import ( + "github.com/fil-forge/ucantone/binding" + "github.com/fil-forge/ucantone/ucan/command" +) + +// Sample returns a usage time series for the space that is the invocation +// subject. The service buckets the series itself, so the caller receives one +// sample per window with no gaps. +var Sample = binding.Bind[*SampleArguments, *SampleOK](command.MustParse("/metrics/sample")) + +const ( + // InvalidRangeErrorName is returned when the requested range is not a + // non-empty interval: To must be after From, and both must be positive Unix + // timestamps. + InvalidRangeErrorName = "InvalidRange" + // InvalidWindowErrorName is returned when the requested window is not a + // positive, representable number of seconds. + InvalidWindowErrorName = "InvalidWindow" + // TooManySamplesErrorName is returned when the range divided by the window + // exceeds the number of samples the service returns in one response. The + // message carries the limit. Callers should widen the window, or split the + // range across several requests. + TooManySamplesErrorName = "TooManySamples" + // UsageUnstableErrorName is returned when the service could not read the + // space's usage consistently, because the space was written to throughout + // the attempt. It is retryable. + UsageUnstableErrorName = "UsageUnstable" + // RangeTooBusyErrorName is returned when the range holds more recorded + // changes than the service will scan in one response. Callers should split + // the range across several requests. + RangeTooBusyErrorName = "RangeTooBusy" +) diff --git a/commands/metrics/types.go b/commands/metrics/types.go new file mode 100644 index 0000000..7629145 --- /dev/null +++ b/commands/metrics/types.go @@ -0,0 +1,70 @@ +// Package metrics defines the usage metering capabilities. The space the +// samples describe is the invocation subject, so it never appears in the +// arguments. +package metrics + +// SampleArguments is the argument shape of `/metrics/sample`. It asks for a +// usage time series for the invocation subject, covering [From, To) in buckets +// of Window seconds. +// +// All three fields are required and are expressed in seconds: From and To are +// Unix timestamps, From inclusive and To exclusive, and Window is a duration. +// Window need not divide To-From evenly — the final bucket is then short and +// ends at To. +// +// The series the service returns is dense: a bucket in which nothing happened +// still yields a sample. Consumers take an unweighted mean over the stored-bytes +// series, so a bucket dropped for want of activity would reweight that mean. +type SampleArguments struct { + // From is the start of the range, inclusive, as a Unix timestamp in seconds. + From int64 `cborgen:"from" dagjsongen:"from"` + // To is the end of the range, exclusive, as a Unix timestamp in seconds. + To int64 `cborgen:"to" dagjsongen:"to"` + // Window is the width of one bucket, in seconds. It must be positive. + Window int64 `cborgen:"window" dagjsongen:"window"` +} + +// SampleOK is the success return for `/metrics/sample`. Samples holds exactly +// one entry per bucket in [From, To), ordered by ascending Timestamp, with no +// gaps. +// +// From, To and Window restate the range the samples cover. From and Window +// always equal the request's. To is the requested To clamped to the service's +// current time: a bucket that has not closed has no value to report, and +// padding one with the latest reading would bias a consumer's average. A +// request whose range lies entirely in the future returns no samples. +type SampleOK struct { + // From is the start of the range covered, inclusive, as a Unix timestamp in + // seconds. + From int64 `cborgen:"from" dagjsongen:"from"` + // To is the end of the range covered, exclusive, as a Unix timestamp in + // seconds. It is the requested To clamped to the service's current time. + To int64 `cborgen:"to" dagjsongen:"to"` + // Window is the width of one bucket, in seconds. + Window int64 `cborgen:"window" dagjsongen:"window"` + // Samples is one entry per bucket, ascending by Timestamp. + Samples []SampleItem `cborgen:"samples" dagjsongen:"samples"` +} + +// SampleItem is one bucket of the series. +// +// Timestamp is the end of the bucket rather than its start. Consumers key a +// sample to the instant its window closes, and a start-of-window timestamp +// lands the value in the neighbouring bucket when several series are collapsed +// onto one grid. +// +// BytesStored is a gauge: the bytes the space holds as of Timestamp, counting +// everything stored and not yet removed before that instant. BytesIngested is a +// flow: the bytes added during the bucket. Removals do not reduce BytesIngested; +// they show up in the next BytesStored. +// +// There is no egress here. Egress is accounted for by the egress tracking +// service, which sees retrieval receipts this service never handles. +type SampleItem struct { + // Timestamp is the end of the bucket, as a Unix timestamp in seconds. + Timestamp int64 `cborgen:"timestamp" dagjsongen:"timestamp"` + // BytesStored is the bytes the space holds at Timestamp. + BytesStored uint64 `cborgen:"bytesStored" dagjsongen:"bytesStored"` + // BytesIngested is the bytes added to the space during the bucket. + BytesIngested uint64 `cborgen:"bytesIngested" dagjsongen:"bytesIngested"` +} From 65165336c658f58b54b20391681c68b163df7799 Mon Sep 17 00:00:00 2001 From: Alan Shaw Date: Tue, 22 Sep 2026 08:17:21 +0100 Subject: [PATCH 2/3] refactor: address review comments --- commands/metrics/types.go | 19 ++++++++++++++----- 1 file changed, 14 insertions(+), 5 deletions(-) diff --git a/commands/metrics/types.go b/commands/metrics/types.go index 7629145..f20559d 100644 --- a/commands/metrics/types.go +++ b/commands/metrics/types.go @@ -9,8 +9,9 @@ package metrics // // All three fields are required and are expressed in seconds: From and To are // Unix timestamps, From inclusive and To exclusive, and Window is a duration. -// Window need not divide To-From evenly — the final bucket is then short and -// ends at To. +// Window need not divide the range evenly, and To may lie in the future; the +// service shortens the range to end at the present and SampleOK reports the +// range actually covered. // // The series the service returns is dense: a bucket in which nothing happened // still yields a sample. Consumers take an unweighted mean over the stored-bytes @@ -30,9 +31,17 @@ type SampleArguments struct { // // From, To and Window restate the range the samples cover. From and Window // always equal the request's. To is the requested To clamped to the service's -// current time: a bucket that has not closed has no value to report, and -// padding one with the latest reading would bias a consumer's average. A -// request whose range lies entirely in the future returns no samples. +// current time, and never earlier than From: the service reports what has +// already happened rather than padding out buckets that lie ahead. A request +// whose range is entirely in the future therefore returns To equal to From and +// no samples. +// +// The last bucket is short whenever Window does not divide To-From, which +// includes the ordinary case of a range running up to the present. A short +// bucket is still a bucket: it carries one sample, timestamped at To. Its +// BytesStored is exact, being a reading taken at that instant, while its +// BytesIngested covers only the part of the window that has elapsed and so +// reads low against a full one. type SampleOK struct { // From is the start of the range covered, inclusive, as a Unix timestamp in // seconds. From 729c9a816140ee7cb97d9293bd2af17e9241e02e Mon Sep 17 00:00:00 2001 From: Alan Shaw Date: Tue, 22 Sep 2026 11:47:22 +0100 Subject: [PATCH 3/3] fix: provider aware metrics --- .../metrics/{cbor_gen.go => cbor_gen.maps.go} | 226 +---------------- commands/metrics/cbor_gen.tuples.go | 140 +++++++++++ commands/metrics/codec.go | 129 ++++++++++ commands/metrics/codec_test.go | 99 ++++++-- commands/metrics/gen/main.go | 30 ++- .../metrics/{json_gen.go => json_gen.maps.go} | 231 +----------------- commands/metrics/json_gen.tuples.go | 145 +++++++++++ commands/metrics/types.go | 42 +++- 8 files changed, 567 insertions(+), 475 deletions(-) rename commands/metrics/{cbor_gen.go => cbor_gen.maps.go} (66%) create mode 100644 commands/metrics/cbor_gen.tuples.go create mode 100644 commands/metrics/codec.go rename commands/metrics/{json_gen.go => json_gen.maps.go} (59%) create mode 100644 commands/metrics/json_gen.tuples.go diff --git a/commands/metrics/cbor_gen.go b/commands/metrics/cbor_gen.maps.go similarity index 66% rename from commands/metrics/cbor_gen.go rename to commands/metrics/cbor_gen.maps.go index 5247a22..b6eff36 100644 --- a/commands/metrics/cbor_gen.go +++ b/commands/metrics/cbor_gen.maps.go @@ -309,7 +309,7 @@ func (t *SampleOK) MarshalCBOR(w io.Writer) error { } } - // t.Samples ([]metrics.SampleItem) (slice) + // t.Samples (metrics.SampleSet) (struct) if len("samples") > 8192 { return xerrors.Errorf("Value in field \"samples\" was too long") } @@ -321,19 +321,9 @@ func (t *SampleOK) MarshalCBOR(w io.Writer) error { return err } - if len(t.Samples) > 8192 { - return xerrors.Errorf("Slice value in field t.Samples was too long") - } - - if err := cw.WriteMajorTypeHeader(cbg.MajArray, uint64(len(t.Samples))); err != nil { + if err := t.Samples.MarshalCBOR(cw); err != nil { return err } - for _, v := range t.Samples { - if err := v.MarshalCBOR(cw); err != nil { - return err - } - - } return nil } @@ -456,220 +446,14 @@ func (t *SampleOK) UnmarshalCBOR(r io.Reader) (err error) { t.Window = int64(extraI) } - // t.Samples ([]metrics.SampleItem) (slice) + // t.Samples (metrics.SampleSet) (struct) case "samples": - maj, extra, err = cr.ReadHeader() - if err != nil { - return err - } - - if extra > 8192 { - return fmt.Errorf("t.Samples: array too large (%d)", extra) - } - - if maj != cbg.MajArray { - return fmt.Errorf("expected cbor array") - } - - if extra > 0 { - t.Samples = make([]SampleItem, extra) - } - - for i := 0; i < int(extra); i++ { - { - var maj byte - var extra uint64 - var err error - _ = maj - _ = extra - _ = err - - { - - if err := t.Samples[i].UnmarshalCBOR(cr); err != nil { - return xerrors.Errorf("unmarshaling t.Samples[i]: %w", err) - } - - } - - } - } - - default: - // Field doesn't exist on this type, so ignore it - if err := cbg.ScanForLinks(r, func(cid.Cid) {}); err != nil { - return err - } - } - } - - return nil -} -func (t *SampleItem) MarshalCBOR(w io.Writer) error { - if t == nil { - _, err := w.Write(cbg.CborNull) - return err - } - - cw := cbg.NewCborWriter(w) - - if _, err := cw.Write([]byte{163}); err != nil { - return err - } - - // t.Timestamp (int64) (int64) - if len("timestamp") > 8192 { - return xerrors.Errorf("Value in field \"timestamp\" was too long") - } - - if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("timestamp"))); err != nil { - return err - } - if _, err := cw.WriteString(string("timestamp")); err != nil { - return err - } - - if t.Timestamp >= 0 { - if err := cw.WriteMajorTypeHeader(cbg.MajUnsignedInt, uint64(t.Timestamp)); err != nil { - return err - } - } else { - if err := cw.WriteMajorTypeHeader(cbg.MajNegativeInt, uint64(-t.Timestamp-1)); err != nil { - return err - } - } - - // t.BytesStored (uint64) (uint64) - if len("bytesStored") > 8192 { - return xerrors.Errorf("Value in field \"bytesStored\" was too long") - } - - if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("bytesStored"))); err != nil { - return err - } - if _, err := cw.WriteString(string("bytesStored")); err != nil { - return err - } - - if err := cw.WriteMajorTypeHeader(cbg.MajUnsignedInt, uint64(t.BytesStored)); err != nil { - return err - } - - // t.BytesIngested (uint64) (uint64) - if len("bytesIngested") > 8192 { - return xerrors.Errorf("Value in field \"bytesIngested\" was too long") - } - - if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("bytesIngested"))); err != nil { - return err - } - if _, err := cw.WriteString(string("bytesIngested")); err != nil { - return err - } - - if err := cw.WriteMajorTypeHeader(cbg.MajUnsignedInt, uint64(t.BytesIngested)); err != nil { - return err - } - - return nil -} - -func (t *SampleItem) UnmarshalCBOR(r io.Reader) (err error) { - *t = SampleItem{} - - cr := cbg.NewCborReader(r) - - maj, extra, err := cr.ReadHeader() - if err != nil { - return err - } - defer func() { - if err == io.EOF { - err = io.ErrUnexpectedEOF - } - }() - - if maj != cbg.MajMap { - return fmt.Errorf("cbor input should be of type map") - } - - if extra > cbg.MaxLength { - return fmt.Errorf("SampleItem: map struct too large (%d)", extra) - } - - n := extra - - nameBuf := make([]byte, 13) - for i := uint64(0); i < n; i++ { - nameLen, ok, err := cbg.ReadFullStringIntoBuf(cr, nameBuf, 8192) - if err != nil { - return err - } - - if !ok { - // Field doesn't exist on this type, so ignore it - if err := cbg.ScanForLinks(cr, func(cid.Cid) {}); err != nil { - return err - } - continue - } - - switch string(nameBuf[:nameLen]) { - // t.Timestamp (int64) (int64) - case "timestamp": - { - maj, extra, err := cr.ReadHeader() - if err != nil { - return err - } - var extraI int64 - switch maj { - case cbg.MajUnsignedInt: - extraI = int64(extra) - if extraI < 0 { - return fmt.Errorf("int64 positive overflow") - } - case cbg.MajNegativeInt: - extraI = int64(extra) - if extraI < 0 { - return fmt.Errorf("int64 negative overflow") - } - extraI = -1 - extraI - default: - return fmt.Errorf("wrong type for int64 field: %d", maj) - } - - t.Timestamp = int64(extraI) - } - // t.BytesStored (uint64) (uint64) - case "bytesStored": - { - maj, extra, err = cr.ReadHeader() - if err != nil { - return err - } - if maj != cbg.MajUnsignedInt { - return fmt.Errorf("wrong type for uint64 field") - } - t.BytesStored = uint64(extra) - - } - // t.BytesIngested (uint64) (uint64) - case "bytesIngested": - - { - - maj, extra, err = cr.ReadHeader() - if err != nil { - return err - } - if maj != cbg.MajUnsignedInt { - return fmt.Errorf("wrong type for uint64 field") + if err := t.Samples.UnmarshalCBOR(cr); err != nil { + return xerrors.Errorf("unmarshaling t.Samples: %w", err) } - t.BytesIngested = uint64(extra) } diff --git a/commands/metrics/cbor_gen.tuples.go b/commands/metrics/cbor_gen.tuples.go new file mode 100644 index 0000000..801a8af --- /dev/null +++ b/commands/metrics/cbor_gen.tuples.go @@ -0,0 +1,140 @@ +//go:build !codegen + +// Code generated by github.com/whyrusleeping/cbor-gen. DO NOT EDIT. + +package metrics + +import ( + "fmt" + "io" + "math" + "sort" + + cid "github.com/ipfs/go-cid" + cbg "github.com/whyrusleeping/cbor-gen" + xerrors "golang.org/x/xerrors" +) + +var _ = xerrors.Errorf +var _ = cid.Undef +var _ = math.E +var _ = sort.Sort + +var lengthBufSampleItem = []byte{131} + +func (t *SampleItem) MarshalCBOR(w io.Writer) error { + if t == nil { + _, err := w.Write(cbg.CborNull) + return err + } + + cw := cbg.NewCborWriter(w) + + if _, err := cw.Write(lengthBufSampleItem); err != nil { + return err + } + + // t.Timestamp (int64) (int64) + if t.Timestamp >= 0 { + if err := cw.WriteMajorTypeHeader(cbg.MajUnsignedInt, uint64(t.Timestamp)); err != nil { + return err + } + } else { + if err := cw.WriteMajorTypeHeader(cbg.MajNegativeInt, uint64(-t.Timestamp-1)); err != nil { + return err + } + } + + // t.BytesStored (uint64) (uint64) + + if err := cw.WriteMajorTypeHeader(cbg.MajUnsignedInt, uint64(t.BytesStored)); err != nil { + return err + } + + // t.BytesIngested (uint64) (uint64) + + if err := cw.WriteMajorTypeHeader(cbg.MajUnsignedInt, uint64(t.BytesIngested)); err != nil { + return err + } + + return nil +} + +func (t *SampleItem) UnmarshalCBOR(r io.Reader) (err error) { + *t = SampleItem{} + + cr := cbg.NewCborReader(r) + + maj, extra, err := cr.ReadHeader() + if err != nil { + return err + } + defer func() { + if err == io.EOF { + err = io.ErrUnexpectedEOF + } + }() + + if maj != cbg.MajArray { + return fmt.Errorf("cbor input should be of type array") + } + + if extra != 3 { + return fmt.Errorf("cbor input had wrong number of fields") + } + + // t.Timestamp (int64) (int64) + { + maj, extra, err := cr.ReadHeader() + if err != nil { + return err + } + var extraI int64 + switch maj { + case cbg.MajUnsignedInt: + extraI = int64(extra) + if extraI < 0 { + return fmt.Errorf("int64 positive overflow") + } + case cbg.MajNegativeInt: + extraI = int64(extra) + if extraI < 0 { + return fmt.Errorf("int64 negative overflow") + } + extraI = -1 - extraI + default: + return fmt.Errorf("wrong type for int64 field: %d", maj) + } + + t.Timestamp = int64(extraI) + } + // t.BytesStored (uint64) (uint64) + + { + + maj, extra, err = cr.ReadHeader() + if err != nil { + return err + } + if maj != cbg.MajUnsignedInt { + return fmt.Errorf("wrong type for uint64 field") + } + t.BytesStored = uint64(extra) + + } + // t.BytesIngested (uint64) (uint64) + + { + + maj, extra, err = cr.ReadHeader() + if err != nil { + return err + } + if maj != cbg.MajUnsignedInt { + return fmt.Errorf("wrong type for uint64 field") + } + t.BytesIngested = uint64(extra) + + } + return nil +} diff --git a/commands/metrics/codec.go b/commands/metrics/codec.go new file mode 100644 index 0000000..a876110 --- /dev/null +++ b/commands/metrics/codec.go @@ -0,0 +1,129 @@ +//go:build !codegen + +package metrics + +import ( + "io" + + jsg "github.com/alanshaw/dag-json-gen" + "github.com/fil-forge/libforge/commands/internal/codec" + "github.com/fil-forge/ucantone/did" + cbg "github.com/whyrusleeping/cbor-gen" + xerrors "golang.org/x/xerrors" +) + +// SampleSet has a hand-written codec because cbor-gen / dag-json-gen do not +// support DID-keyed maps. The CBOR and DAG-JSON primitives live in the shared +// internal codec package. + +func (t SampleSet) MarshalCBOR(w io.Writer) error { + cw := cbg.NewCborWriter(w) + if err := codec.WriteCborMapHeader(cw, len(t.Entries)); err != nil { + return err + } + for _, k := range codec.SortedDIDs(t.Entries) { + if err := codec.WriteCborString(cw, k.String()); err != nil { + return err + } + samples := t.Entries[k] + if err := codec.WriteCborArrayHeader(cw, len(samples)); err != nil { + return err + } + for i := range samples { + if err := samples[i].MarshalCBOR(cw); err != nil { + return err + } + } + } + return nil +} + +func (t *SampleSet) UnmarshalCBOR(r io.Reader) error { + cr := cbg.NewCborReader(r) + n, err := codec.ReadCborMapHeader(cr) + if err != nil { + return err + } + m := make(map[did.DID][]SampleItem, n) + for i := uint64(0); i < n; i++ { + ks, err := codec.ReadCborString(cr) + if err != nil { + return err + } + k, err := did.Parse(ks) + if err != nil { + return xerrors.Errorf("parsing provider did %q: %w", ks, err) + } + sn, err := codec.ReadCborArrayHeader(cr) + if err != nil { + return err + } + samples := make([]SampleItem, sn) + for j := uint64(0); j < sn; j++ { + if err := samples[j].UnmarshalCBOR(cr); err != nil { + return err + } + } + m[k] = samples + } + *t = SampleSet{Entries: m} + return nil +} + +func (t SampleSet) MarshalDagJSON(w io.Writer) error { + jw := jsg.NewDagJsonWriter(w) + if err := jw.WriteObjectOpen(); err != nil { + return err + } + for i, k := range codec.SortedDIDs(t.Entries) { + if err := codec.WriteJSONKey(jw, k.String(), i > 0); err != nil { + return err + } + if err := jw.WriteArrayOpen(); err != nil { + return err + } + samples := t.Entries[k] + for j := range samples { + if j > 0 { + if err := jw.WriteComma(); err != nil { + return err + } + } + if err := samples[j].MarshalDagJSON(jw); err != nil { + return err + } + } + if err := jw.WriteArrayClose(); err != nil { + return err + } + } + return jw.WriteObjectClose() +} + +func (t *SampleSet) UnmarshalDagJSON(r io.Reader) error { + jr := jsg.NewDagJsonReader(r) + m := map[did.DID][]SampleItem{} + err := codec.ReadJSONObject(jr, func(ks string) error { + k, err := did.Parse(ks) + if err != nil { + return xerrors.Errorf("parsing provider did %q: %w", ks, err) + } + samples, err := codec.ReadJSONArray(jr, func() (SampleItem, error) { + var s SampleItem + if err := s.UnmarshalDagJSON(jr); err != nil { + return SampleItem{}, err + } + return s, nil + }) + if err != nil { + return err + } + m[k] = samples + return nil + }) + if err != nil { + return err + } + *t = SampleSet{Entries: m} + return nil +} diff --git a/commands/metrics/codec_test.go b/commands/metrics/codec_test.go index fc95590..1c3b328 100644 --- a/commands/metrics/codec_test.go +++ b/commands/metrics/codec_test.go @@ -9,6 +9,7 @@ import ( "testing" "github.com/fil-forge/libforge/commands/metrics" + "github.com/fil-forge/ucantone/did" "github.com/stretchr/testify/require" ) @@ -45,35 +46,97 @@ func TestSampleArgumentsRoundTrip(t *testing.T) { require.Equal(t, `{"from":1700000000,"to":1700003600,"window":3600}`, roundTrip(t, in)) } +// provider is a fixed DID so the expected DAG-JSON below is stable. +var provider = did.MustParse("did:web:provider.example") + func TestSampleOKRoundTrip(t *testing.T) { in := &metrics.SampleOK{ From: 1700000000, To: 1700007200, Window: 3600, - Samples: []metrics.SampleItem{ - {Timestamp: 1700003600, BytesStored: 1024, BytesIngested: 1024}, - {Timestamp: 1700007200, BytesStored: 512, BytesIngested: 0}, - }, + Samples: metrics.SampleSet{Entries: map[did.DID][]metrics.SampleItem{ + provider: { + {Timestamp: 1700003600, BytesStored: 1024, BytesIngested: 1024}, + {Timestamp: 1700007200, BytesStored: 512, BytesIngested: 0}, + }, + }}, } + // SampleItem is tuple encoded: [timestamp, bytesStored, bytesIngested]. require.Equal(t, - `{"from":1700000000,"samples":[`+ - `{"bytesIngested":1024,"bytesStored":1024,"timestamp":1700003600},`+ - `{"bytesIngested":0,"bytesStored":512,"timestamp":1700007200}`+ - `],"to":1700007200,"window":3600}`, + `{"from":1700000000,"samples":{"did:web:provider.example":[`+ + `[1700003600,1024,1024],`+ + `[1700007200,512,0]`+ + `]},"to":1700007200,"window":3600}`, roundTrip(t, in)) } -// A range entirely in the future carries no samples, so the empty series has to -// survive both codecs. -func TestSampleOKEmptySeriesRoundTrip(t *testing.T) { +// A space provisioned with several providers carries a series each, on one +// shared bucket grid. The keys sort, so the encoding is deterministic. +func TestSampleOKSeveralProvidersRoundTrip(t *testing.T) { + a := did.MustParse("did:web:a.example") + b := did.MustParse("did:web:b.example") + in := &metrics.SampleOK{ + From: 1700000000, + To: 1700003600, + Window: 3600, + Samples: metrics.SampleSet{Entries: map[did.DID][]metrics.SampleItem{ + b: {{Timestamp: 1700003600, BytesStored: 2, BytesIngested: 2}}, + a: {{Timestamp: 1700003600, BytesStored: 1, BytesIngested: 1}}, + }}, + } + require.Equal(t, + `{"from":1700000000,"samples":{`+ + `"did:web:a.example":[[1700003600,1,1]],`+ + `"did:web:b.example":[[1700003600,2,2]]`+ + `},"to":1700003600,"window":3600}`, + roundTrip(t, in)) +} + +// A range entirely in the future carries no samples. An absent set and an +// empty one are the same thing on the wire, so these assert the encoding and +// the decoded content rather than round-trip identity. +func TestSampleOKEmptySetRoundTrip(t *testing.T) { in := &metrics.SampleOK{From: 1700000000, To: 1700000000, Window: 3600} - require.Equal(t, `{"from":1700000000,"samples":[],"to":1700000000,"window":3600}`, roundTrip(t, in)) + + var jb bytes.Buffer + require.NoError(t, in.MarshalDagJSON(&jb)) + require.Equal(t, `{"from":1700000000,"samples":{},"to":1700000000,"window":3600}`, jb.String()) + + var fromJSON metrics.SampleOK + require.NoError(t, fromJSON.UnmarshalDagJSON(bytes.NewReader(jb.Bytes()))) + require.Empty(t, fromJSON.Samples.Entries) var cb bytes.Buffer require.NoError(t, in.MarshalCBOR(&cb)) - var out metrics.SampleOK - require.NoError(t, out.UnmarshalCBOR(bytes.NewReader(cb.Bytes()))) - require.Empty(t, out.Samples) + var fromCBOR metrics.SampleOK + require.NoError(t, fromCBOR.UnmarshalCBOR(bytes.NewReader(cb.Bytes()))) + require.Empty(t, fromCBOR.Samples.Entries) +} + +// A provider present with nothing to report keeps its key and an empty series. +func TestSampleOKEmptySeriesRoundTrip(t *testing.T) { + in := &metrics.SampleOK{ + From: 1700000000, To: 1700000000, Window: 3600, + Samples: metrics.SampleSet{Entries: map[did.DID][]metrics.SampleItem{provider: {}}}, + } + + var jb bytes.Buffer + require.NoError(t, in.MarshalDagJSON(&jb)) + require.Equal(t, + `{"from":1700000000,"samples":{"did:web:provider.example":[]},"to":1700000000,"window":3600}`, + jb.String()) + + var fromJSON metrics.SampleOK + require.NoError(t, fromJSON.UnmarshalDagJSON(bytes.NewReader(jb.Bytes()))) + require.Len(t, fromJSON.Samples.Entries, 1) + require.Empty(t, fromJSON.Samples.Entries[provider]) + + var cb bytes.Buffer + require.NoError(t, in.MarshalCBOR(&cb)) + var fromCBOR metrics.SampleOK + require.NoError(t, fromCBOR.UnmarshalCBOR(bytes.NewReader(cb.Bytes()))) + require.Len(t, fromCBOR.Samples.Entries, 1) + require.Empty(t, fromCBOR.Samples.Entries[provider]) } // 768 samples is a 32 day range at hourly granularity, the largest series @@ -85,20 +148,22 @@ func TestSampleOKLargeSeriesRoundTrip(t *testing.T) { count = 768 ) in := &metrics.SampleOK{From: 1700000000, Window: hour} + samples := make([]metrics.SampleItem, 0, count) for i := range count { - in.Samples = append(in.Samples, metrics.SampleItem{ + samples = append(samples, metrics.SampleItem{ Timestamp: in.From + int64(i+1)*hour, BytesStored: uint64(i) * 1 << 20, BytesIngested: 1 << 20, }) } + in.Samples = metrics.SampleSet{Entries: map[did.DID][]metrics.SampleItem{provider: samples}} in.To = in.From + count*hour var cb bytes.Buffer require.NoError(t, in.MarshalCBOR(&cb)) var out metrics.SampleOK require.NoError(t, out.UnmarshalCBOR(bytes.NewReader(cb.Bytes()))) - require.Len(t, out.Samples, count) + require.Len(t, out.Samples.Entries[provider], count) require.Equal(t, *in, out) t.Logf("%d samples encode to %d bytes of CBOR", count, cb.Len()) } diff --git a/commands/metrics/gen/main.go b/commands/metrics/gen/main.go index c871b91..1a03f15 100644 --- a/commands/metrics/gen/main.go +++ b/commands/metrics/gen/main.go @@ -23,21 +23,37 @@ func tag(path string) { } func main() { - models := []any{ + // SampleSet has a hand-written codec (see codec.go) because cbor-gen / + // dag-json-gen do not support DID-keyed maps. + mapModels := []any{ metrics.SampleArguments{}, metrics.SampleOK{}, + } + // SampleItem is tuple encoded: it repeats once per bucket, and a map would + // repeat its three key names with it. + tupleModels := []any{ metrics.SampleItem{}, } const ( - cborFile = "../cbor_gen.go" - jsonFile = "../json_gen.go" + cborTuples = "../cbor_gen.tuples.go" + cborMaps = "../cbor_gen.maps.go" + jsonTuples = "../json_gen.tuples.go" + jsonMaps = "../json_gen.maps.go" ) - if err := cbg.WriteMapEncodersToFile(cborFile, "metrics", models...); err != nil { + if err := cbg.WriteTupleEncodersToFile(cborTuples, "metrics", tupleModels...); err != nil { + panic(err) + } + if err := cbg.WriteMapEncodersToFile(cborMaps, "metrics", mapModels...); err != nil { + panic(err) + } + if err := jsg.WriteTupleEncodersToFile(jsonTuples, "metrics", tupleModels...); err != nil { panic(err) } - if err := jsg.WriteMapEncodersToFile(jsonFile, "metrics", models...); err != nil { + if err := jsg.WriteMapEncodersToFile(jsonMaps, "metrics", mapModels...); err != nil { panic(err) } - tag(cborFile) - tag(jsonFile) + tag(cborTuples) + tag(cborMaps) + tag(jsonTuples) + tag(jsonMaps) } diff --git a/commands/metrics/json_gen.go b/commands/metrics/json_gen.maps.go similarity index 59% rename from commands/metrics/json_gen.go rename to commands/metrics/json_gen.maps.go index 264f70e..5af6d89 100644 --- a/commands/metrics/json_gen.go +++ b/commands/metrics/json_gen.maps.go @@ -219,7 +219,7 @@ func (t *SampleOK) MarshalDagJSON(w io.Writer) error { } } - // t.Samples ([]metrics.SampleItem) (slice) + // t.Samples (metrics.SampleSet) (struct) if len("samples") > 8192 { return fmt.Errorf("string in field \"samples\" was too long") } @@ -229,27 +229,9 @@ func (t *SampleOK) MarshalDagJSON(w io.Writer) error { if err := jw.WriteObjectColon(); err != nil { return err } - if len(t.Samples) > 8192 { - return fmt.Errorf("slice value in field t.Samples was too long") + if err := t.Samples.MarshalDagJSON(jw); err != nil { + return fmt.Errorf("marshaling field t.Samples: %w", err) } - - if err := jw.WriteArrayOpen(); err != nil { - return fmt.Errorf("writing array open for field t.Samples: %w", err) - } - for i, v := range t.Samples { - if i > 0 { - if err := jw.WriteComma(); err != nil { - return fmt.Errorf("writing comma for field t.Samples: %w", err) - } - } - if err := v.MarshalDagJSON(jw); err != nil { - return fmt.Errorf("marshaling field v: %w", err) - } - } - if err := jw.WriteArrayClose(); err != nil { - return fmt.Errorf("writing array close for field t.Samples: %w", err) - } - written = true if written { if err := jw.WriteComma(); err != nil { @@ -345,46 +327,11 @@ func (t *SampleOK) UnmarshalDagJSON(r io.Reader) (err error) { } - // t.Samples ([]metrics.SampleItem) (slice) + // t.Samples (metrics.SampleSet) (struct) case "samples": - { - - if err := jr.ReadArrayOpen(); err != nil { - return fmt.Errorf("reading array open for field t.Samples: %w", err) - } - - close, err := jr.PeekArrayClose() - if err != nil { - return fmt.Errorf("peeking array close for field t.Samples: %w", err) - } - if close { - if err := jr.ReadArrayClose(); err != nil { - return fmt.Errorf("reading array close for field t.Samples: %w", err) - } - - } else { - for i := 0; i < 8192; i++ { - item := make([]SampleItem, 1) - - if err := item[0].UnmarshalDagJSON(jr); err != nil { - return fmt.Errorf("unmarshaling item[0]: %w", err) - } - - t.Samples = append(t.Samples, item[0]) - - close, err := jr.ReadArrayCloseOrComma() - if err != nil { - return fmt.Errorf("reading array close or comma for field t.Samples: %w", err) - } - if close { - break - } - if i == 8192-1 { - return fmt.Errorf("reading array for field t.Samples: slice too large") - } - } - } + if err := t.Samples.UnmarshalDagJSON(jr); err != nil { + return fmt.Errorf("unmarshaling t.Samples: %w", err) } // t.To (int64) (int64) @@ -432,169 +379,3 @@ func (t *SampleOK) UnmarshalDagJSON(r io.Reader) (err error) { return nil } -func (t *SampleItem) MarshalDagJSON(w io.Writer) error { - jw := jsg.NewDagJsonWriter(w) - if t == nil { - err := jw.WriteNull() - return err - } - if err := jw.WriteObjectOpen(); err != nil { - return err - } - written := false - - // t.BytesIngested (uint64) (uint64) - if len("bytesIngested") > 8192 { - return fmt.Errorf("string in field \"bytesIngested\" was too long") - } - if err := jw.WriteString(string("bytesIngested")); err != nil { - return fmt.Errorf("writing string for field \"bytesIngested\": %w", err) - } - if err := jw.WriteObjectColon(); err != nil { - return err - } - - if err := jw.WriteUint64(uint64(t.BytesIngested)); err != nil { - return fmt.Errorf("writing uint64 for field t.BytesIngested: %w", err) - } - - written = true - if written { - if err := jw.WriteComma(); err != nil { - return err - } - } - - // t.BytesStored (uint64) (uint64) - if len("bytesStored") > 8192 { - return fmt.Errorf("string in field \"bytesStored\" was too long") - } - if err := jw.WriteString(string("bytesStored")); err != nil { - return fmt.Errorf("writing string for field \"bytesStored\": %w", err) - } - if err := jw.WriteObjectColon(); err != nil { - return err - } - - if err := jw.WriteUint64(uint64(t.BytesStored)); err != nil { - return fmt.Errorf("writing uint64 for field t.BytesStored: %w", err) - } - - written = true - if written { - if err := jw.WriteComma(); err != nil { - return err - } - } - - // t.Timestamp (int64) (int64) - if len("timestamp") > 8192 { - return fmt.Errorf("string in field \"timestamp\" was too long") - } - if err := jw.WriteString(string("timestamp")); err != nil { - return fmt.Errorf("writing string for field \"timestamp\": %w", err) - } - if err := jw.WriteObjectColon(); err != nil { - return err - } - - if err := jw.WriteInt64(int64(t.Timestamp)); err != nil { - return fmt.Errorf("writing int64 for field t.Timestamp: %w", err) - } - - if err := jw.WriteObjectClose(); err != nil { - return err - } - return nil -} -func (t *SampleItem) UnmarshalDagJSON(r io.Reader) (err error) { - *t = SampleItem{} - - jr := jsg.NewDagJsonReader(r) - defer func() { - if err == io.EOF { - err = io.ErrUnexpectedEOF - } - }() - if err := jr.ReadObjectOpen(); err != nil { - return fmt.Errorf("reading object open for SampleItem: %w", err) - } - close, err := jr.PeekObjectClose() - if err != nil { - return fmt.Errorf("peeking object close for SampleItem: %w", err) - } - if close { - if err := jr.ReadObjectClose(); err != nil { - return fmt.Errorf("reading object close for SampleItem: %w", err) - } - } else { - for i := uint64(0); i < 8192; i++ { - name, err := jr.ReadString(8192) - if err != nil { - if errors.Is(err, jsg.ErrLimitExceeded) { - return fmt.Errorf("reading string for field SampleItem: string too large") - } - return fmt.Errorf("reading string for field SampleItem: %w", err) - } - if err := jr.ReadObjectColon(); err != nil { - return fmt.Errorf("reading object colon for field SampleItem: %w", err) - } - switch name { - - // t.BytesIngested (uint64) (uint64) - case "bytesIngested": - { - - nval, err := jr.ReadNumberAsUint64() - if err != nil { - return fmt.Errorf("reading uint64 for field t.BytesIngested: %w", err) - } - t.BytesIngested = uint64(nval) - - } - - // t.BytesStored (uint64) (uint64) - case "bytesStored": - { - - nval, err := jr.ReadNumberAsUint64() - if err != nil { - return fmt.Errorf("reading uint64 for field t.BytesStored: %w", err) - } - t.BytesStored = uint64(nval) - - } - - // t.Timestamp (int64) (int64) - case "timestamp": - { - - nval, err := jr.ReadNumberAsInt64() - if err != nil { - return fmt.Errorf("reading int64 for field t.Timestamp: %w", err) - } - t.Timestamp = int64(nval) - - } - default: - // Field doesn't exist on this type, so ignore it - if err := jr.DiscardType(); err != nil { - return fmt.Errorf("ignoring field %s for SampleItem: %w", name, err) - } - } - - close, err := jr.ReadObjectCloseOrComma() - if err != nil { - return fmt.Errorf("reading object close or comma for field SampleItem: %w", err) - } - if close { - break - } - if i == 8192-1 { - return fmt.Errorf("map too large for SampleItem") - } - } - } - - return nil -} diff --git a/commands/metrics/json_gen.tuples.go b/commands/metrics/json_gen.tuples.go new file mode 100644 index 0000000..69b0e66 --- /dev/null +++ b/commands/metrics/json_gen.tuples.go @@ -0,0 +1,145 @@ +//go:build !codegen + +// Code generated by github.com/alanshaw/dag-json-gen. DO NOT EDIT. + +package metrics + +import ( + "errors" + "fmt" + "io" + "math" + "sort" + + jsg "github.com/alanshaw/dag-json-gen" + cid "github.com/ipfs/go-cid" +) + +var _ = cid.Undef +var _ = math.E +var _ = sort.Sort +var _ = errors.Is + +func (t *SampleItem) MarshalDagJSON(w io.Writer) error { + jw := jsg.NewDagJsonWriter(w) + if t == nil { + err := jw.WriteNull() + return err + } + if err := jw.WriteArrayOpen(); err != nil { + return fmt.Errorf("writing array open for field SampleItem: %w", err) + } + + // t.Timestamp (int64) (int64) + + if err := jw.WriteInt64(int64(t.Timestamp)); err != nil { + return fmt.Errorf("writing int64 for field t.Timestamp: %w", err) + } + + if err := jw.WriteComma(); err != nil { + return fmt.Errorf("writing comma for field BytesStored: %w", err) + } + + // t.BytesStored (uint64) (uint64) + + if err := jw.WriteUint64(uint64(t.BytesStored)); err != nil { + return fmt.Errorf("writing uint64 for field t.BytesStored: %w", err) + } + + if err := jw.WriteComma(); err != nil { + return fmt.Errorf("writing comma for field BytesIngested: %w", err) + } + + // t.BytesIngested (uint64) (uint64) + + if err := jw.WriteUint64(uint64(t.BytesIngested)); err != nil { + return fmt.Errorf("writing uint64 for field t.BytesIngested: %w", err) + } + + if err := jw.WriteArrayClose(); err != nil { + return fmt.Errorf("writing array close for field SampleItem: %w", err) + } + return nil +} + +func (t *SampleItem) UnmarshalDagJSON(r io.Reader) (err error) { + *t = SampleItem{} + + jr := jsg.NewDagJsonReader(r) + defer func() { + if err == io.EOF { + err = io.ErrUnexpectedEOF + } + }() + if err := jr.ReadArrayOpen(); err != nil { + return fmt.Errorf("reading array open for field SampleItem: %w", err) + } + close, err := jr.PeekArrayClose() + if err != nil { + return fmt.Errorf("peeking array close for field SampleItem: %w", err) + } + if close { + if err := jr.ReadArrayClose(); err != nil { + return fmt.Errorf("reading array close for field SampleItem: %w", err) + } + } else { + + // t.Timestamp (int64) (int64) + + { + + nval, err := jr.ReadNumberAsInt64() + if err != nil { + return fmt.Errorf("reading int64 for field t.Timestamp: %w", err) + } + t.Timestamp = int64(nval) + + } + { + close, err := jr.ReadArrayCloseOrComma() + if err != nil { + return fmt.Errorf("reading array close or comma for field SampleItem: %w", err) + } + if close { + return fmt.Errorf("json input has too few fields 1 < 3") + } + } + + // t.BytesStored (uint64) (uint64) + + { + + nval, err := jr.ReadNumberAsUint64() + if err != nil { + return fmt.Errorf("reading uint64 for field t.BytesStored: %w", err) + } + t.BytesStored = uint64(nval) + + } + { + close, err := jr.ReadArrayCloseOrComma() + if err != nil { + return fmt.Errorf("reading array close or comma for field SampleItem: %w", err) + } + if close { + return fmt.Errorf("json input has too few fields 2 < 3") + } + } + + // t.BytesIngested (uint64) (uint64) + + { + + nval, err := jr.ReadNumberAsUint64() + if err != nil { + return fmt.Errorf("reading uint64 for field t.BytesIngested: %w", err) + } + t.BytesIngested = uint64(nval) + + } + if err := jr.ReadArrayClose(); err != nil { + return fmt.Errorf("reading array close for field SampleItem: %w", err) + } + } + return nil +} diff --git a/commands/metrics/types.go b/commands/metrics/types.go index f20559d..5a192b9 100644 --- a/commands/metrics/types.go +++ b/commands/metrics/types.go @@ -1,8 +1,15 @@ // Package metrics defines the usage metering capabilities. The space the // samples describe is the invocation subject, so it never appears in the // arguments. +// +// SampleSet is a DID-keyed map. cbor-gen / dag-json-gen only generate +// string-keyed maps, so it has a hand-written codec (see codec.go) and is +// defined as a struct wrapping the map so that structs embedding it can still +// be generated as usual. package metrics +import "github.com/fil-forge/ucantone/did" + // SampleArguments is the argument shape of `/metrics/sample`. It asks for a // usage time series for the invocation subject, covering [From, To) in buckets // of Window seconds. @@ -25,9 +32,9 @@ type SampleArguments struct { Window int64 `cborgen:"window" dagjsongen:"window"` } -// SampleOK is the success return for `/metrics/sample`. Samples holds exactly -// one entry per bucket in [From, To), ordered by ascending Timestamp, with no -// gaps. +// SampleOK is the success return for `/metrics/sample`. Each series in Samples +// holds exactly one entry per bucket in [From, To), ordered by ascending +// Timestamp, with no gaps. // // From, To and Window restate the range the samples cover. From and Window // always equal the request's. To is the requested To clamped to the service's @@ -51,12 +58,37 @@ type SampleOK struct { To int64 `cborgen:"to" dagjsongen:"to"` // Window is the width of one bucket, in seconds. Window int64 `cborgen:"window" dagjsongen:"window"` - // Samples is one entry per bucket, ascending by Timestamp. - Samples []SampleItem `cborgen:"samples" dagjsongen:"samples"` + // Samples is the series recorded against each of the space's storage + // providers. + Samples SampleSet `cborgen:"samples" dagjsongen:"samples"` +} + +// SampleSet maps a storage provider DID to the series recorded against it. +// Every series shares one bucket grid: the same count, the same timestamps, +// covering [From, To). +// +// Usage is recorded per provider, so a space provisioned with several holds +// several series. They describe the same stored bytes from each provider's +// side rather than parts of a whole, so a consumer picks the provider it cares +// about; adding them together would count the same bytes once per provider. +// +// It is a struct wrapping the map (rather than a named map type) so that +// structs embedding it can be generated by cbor-gen / dag-json-gen, which +// delegate to a struct field's MarshalCBOR but cannot generate a DID-keyed map +// inline. On the wire it encodes as the bare map, not as an object with an +// "entries" key. +type SampleSet struct { + Entries map[did.DID][]SampleItem } // SampleItem is one bucket of the series. // +// It is tuple encoded, as [Timestamp, BytesStored, BytesIngested], because it +// repeats once per bucket and a map would carry its three key names along with +// it — a 32 day hourly series is about a third the size this way. The order of +// the fields is therefore part of the wire format: a field may be appended, +// but none may be reordered or removed. +// // Timestamp is the end of the bucket rather than its start. Consumers key a // sample to the instant its window closes, and a start-of-window timestamp // lands the value in the neighbouring bucket when several series are collapsed