aboutsummaryrefslogtreecommitdiffstats
path: root/library/go/core/metrics/solomon/stream.go
blob: 7cf6d7006498d6e91559aa782f818a06f8f8af08 (plain) (blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
package solomon

import (
	"context"
	"encoding/json"
	"io"

	"github.com/ydb-platform/ydb/library/go/core/xerrors"
)

const HeaderSize = 24

type StreamFormat string

func (r *Registry) StreamJSON(ctx context.Context, w io.Writer) (written int, err error) {
	cw := newCompressedWriter(w, CompressionNone)

	if ctx.Err() != nil {
		return written, xerrors.Errorf("streamJSON context error: %w", ctx.Err())
	}
	_, err = cw.Write([]byte("{\"metrics\":["))
	if err != nil {
		return written, xerrors.Errorf("write metrics failed: %w", err)
	}

	first := true
	r.metrics.Range(func(_, s interface{}) bool {
		if ctx.Err() != nil {
			err = xerrors.Errorf("streamJSON context error: %w", ctx.Err())
			return false
		}

		// write trailing comma
		if !first {
			_, err = cw.Write([]byte(","))
			if err != nil {
				err = xerrors.Errorf("write metrics failed: %w", err)
				return false
			}
		}

		var b []byte

		b, err = json.Marshal(s)
		if err != nil {
			err = xerrors.Errorf("marshal metric failed: %w", err)
			return false
		}

		// write metric json
		_, err = cw.Write(b)
		if err != nil {
			err = xerrors.Errorf("write metric failed: %w", err)
			return false
		}

		first = false
		return true
	})
	if err != nil {
		return written, err
	}

	if ctx.Err() != nil {
		return written, xerrors.Errorf("streamJSON context error: %w", ctx.Err())
	}
	_, err = cw.Write([]byte("]}"))
	if err != nil {
		return written, xerrors.Errorf("write metrics failed: %w", err)
	}

	if ctx.Err() != nil {
		return written, xerrors.Errorf("streamJSON context error: %w", ctx.Err())
	}
	err = cw.Close()
	if err != nil {
		return written, xerrors.Errorf("close failed: %w", err)
	}

	return cw.(*noCompressionWriteCloser).written, nil
}

func (r *Registry) StreamSpack(ctx context.Context, w io.Writer, compression CompressionType) (int, error) {
	metrics, err := r.Gather()
	if err != nil {
		return 0, err
	}
	return NewSpackEncoder(ctx, compression, metrics).Encode(w)
}