|
| 1 | +package main |
| 2 | + |
| 3 | +import ( |
| 4 | + "encoding/json" |
| 5 | + "flag" |
| 6 | + "fmt" |
| 7 | + "io" |
| 8 | + "os" |
| 9 | + "time" |
| 10 | + |
| 11 | + "github.com/gogo/protobuf/proto" |
| 12 | + "github.com/golang/snappy" |
| 13 | + "github.com/prometheus/prometheus/prompb" |
| 14 | +) |
| 15 | + |
| 16 | +// PrometheusRecord is the ndjson-encoded format used for transporting metrics through firehose |
| 17 | +type PrometheusRecord struct { |
| 18 | + Body []byte `json:"b"` |
| 19 | +} |
| 20 | + |
| 21 | +// TimeSeries represents a decoded Prometheus time series in a more readable format |
| 22 | +type TimeSeries struct { |
| 23 | + Labels map[string]string `json:"labels"` |
| 24 | + Timestamps []int64 `json:"timestamps"` |
| 25 | + Values []float64 `json:"values"` |
| 26 | +} |
| 27 | + |
| 28 | +// WriteRequestJSON is a more readable representation of prompb.WriteRequest |
| 29 | +type WriteRequestJSON struct { |
| 30 | + Timeseries []TimeSeries `json:"timeseries"` |
| 31 | +} |
| 32 | + |
| 33 | +// decodePrompbWriteReq decodes the wrapped prompb.WriteRequest |
| 34 | +func decodePrompbWriteReq(record *PrometheusRecord) (*prompb.WriteRequest, error) { |
| 35 | + // Decompress the snappy-compressed data |
| 36 | + data, err := snappy.Decode(nil, record.Body) |
| 37 | + if err != nil { |
| 38 | + return nil, fmt.Errorf("snappy decode error: %w", err) |
| 39 | + } |
| 40 | + |
| 41 | + // Unmarshal the protobuf message |
| 42 | + var req prompb.WriteRequest |
| 43 | + if err := proto.Unmarshal(data, &req); err != nil { |
| 44 | + return nil, fmt.Errorf("protobuf unmarshal error: %w", err) |
| 45 | + } |
| 46 | + |
| 47 | + return &req, nil |
| 48 | +} |
| 49 | + |
| 50 | +// convertToReadableJSON converts a prompb.WriteRequest to a more readable JSON structure |
| 51 | +func convertToReadableJSON(wreq *prompb.WriteRequest) *WriteRequestJSON { |
| 52 | + result := &WriteRequestJSON{ |
| 53 | + Timeseries: make([]TimeSeries, len(wreq.Timeseries)), |
| 54 | + } |
| 55 | + |
| 56 | + for i, ts := range wreq.Timeseries { |
| 57 | + // Convert labels |
| 58 | + labels := make(map[string]string) |
| 59 | + for _, label := range ts.Labels { |
| 60 | + labels[label.Name] = label.Value |
| 61 | + } |
| 62 | + |
| 63 | + // Extract timestamps and values |
| 64 | + timestamps := make([]int64, len(ts.Samples)) |
| 65 | + values := make([]float64, len(ts.Samples)) |
| 66 | + for j, sample := range ts.Samples { |
| 67 | + timestamps[j] = sample.Timestamp |
| 68 | + values[j] = sample.Value |
| 69 | + } |
| 70 | + |
| 71 | + result.Timeseries[i] = TimeSeries{ |
| 72 | + Labels: labels, |
| 73 | + Timestamps: timestamps, |
| 74 | + Values: values, |
| 75 | + } |
| 76 | + } |
| 77 | + |
| 78 | + return result |
| 79 | +} |
| 80 | + |
| 81 | +// humanReadableTime converts a Prometheus timestamp to a human-readable format |
| 82 | +func humanReadableTime(timestamp int64) string { |
| 83 | + // Prometheus uses milliseconds since epoch |
| 84 | + return time.Unix(timestamp/1000, (timestamp%1000)*1000000).Format(time.RFC3339Nano) |
| 85 | +} |
| 86 | + |
| 87 | +// streamingJSONDecoder reads and processes a stream of JSON objects without requiring them to be newline-delimited |
| 88 | +type streamingJSONDecoder struct { |
| 89 | + decoder *json.Decoder |
| 90 | + count int |
| 91 | +} |
| 92 | + |
| 93 | +func newStreamingJSONDecoder(r io.Reader) *streamingJSONDecoder { |
| 94 | + decoder := json.NewDecoder(r) |
| 95 | + // Configure the decoder to support streams of concatenated JSON objects |
| 96 | + decoder.UseNumber() |
| 97 | + return &streamingJSONDecoder{ |
| 98 | + decoder: decoder, |
| 99 | + count: 0, |
| 100 | + } |
| 101 | +} |
| 102 | + |
| 103 | +func (s *streamingJSONDecoder) next() (*PrometheusRecord, error) { |
| 104 | + var record PrometheusRecord |
| 105 | + if err := s.decoder.Decode(&record); err != nil { |
| 106 | + return nil, err |
| 107 | + } |
| 108 | + s.count++ |
| 109 | + return &record, nil |
| 110 | +} |
| 111 | + |
| 112 | +func main() { |
| 113 | + inputFile := flag.String("input", "", "Input file containing PrometheusRecord entries (one per line)") |
| 114 | + outputFile := flag.String("output", "", "Output file for JSON results (default: stdout)") |
| 115 | + prettyPrint := flag.Bool("pretty", true, "Enable pretty-printing of JSON output") |
| 116 | + humanTime := flag.Bool("human-time", false, "Show human-readable timestamps in output") |
| 117 | + flag.Parse() |
| 118 | + |
| 119 | + if *inputFile == "" { |
| 120 | + fmt.Println("Error: Input file is required") |
| 121 | + flag.Usage() |
| 122 | + os.Exit(1) |
| 123 | + } |
| 124 | + |
| 125 | + // Open input file |
| 126 | + file, err := os.Open(*inputFile) |
| 127 | + if err != nil { |
| 128 | + fmt.Printf("Error opening input file: %v\n", err) |
| 129 | + os.Exit(1) |
| 130 | + } |
| 131 | + defer file.Close() |
| 132 | + |
| 133 | + // Prepare output |
| 134 | + var output *os.File |
| 135 | + if *outputFile == "" { |
| 136 | + output = os.Stdout |
| 137 | + } else { |
| 138 | + output, err = os.Create(*outputFile) |
| 139 | + if err != nil { |
| 140 | + fmt.Printf("Error creating output file: %v\n", err) |
| 141 | + os.Exit(1) |
| 142 | + } |
| 143 | + defer output.Close() |
| 144 | + } |
| 145 | + |
| 146 | + // Create JSON stream decoder |
| 147 | + decoder := newStreamingJSONDecoder(file) |
| 148 | + |
| 149 | + // Process each JSON object in the stream |
| 150 | + for { |
| 151 | + record, err := decoder.next() |
| 152 | + if err == io.EOF { |
| 153 | + break |
| 154 | + } |
| 155 | + if err != nil { |
| 156 | + fmt.Printf("Error parsing JSON object #%d: %v\n", decoder.count, err) |
| 157 | + continue |
| 158 | + } |
| 159 | + |
| 160 | + // Decode the PrometheusRecord |
| 161 | + wreq, err := decodePrompbWriteReq(record) |
| 162 | + if err != nil { |
| 163 | + fmt.Printf("Error decoding JSON object #%d: %v\n", decoder.count, err) |
| 164 | + continue |
| 165 | + } |
| 166 | + |
| 167 | + // Convert to our more readable format |
| 168 | + jsonStruct := convertToReadableJSON(wreq) |
| 169 | + |
| 170 | + // Apply human-readable time conversion if requested |
| 171 | + if *humanTime { |
| 172 | + for i := range jsonStruct.Timeseries { |
| 173 | + humanTimes := make([]string, len(jsonStruct.Timeseries[i].Timestamps)) |
| 174 | + for j, ts := range jsonStruct.Timeseries[i].Timestamps { |
| 175 | + humanTimes[j] = humanReadableTime(ts) |
| 176 | + } |
| 177 | + // We need to output this differently, so create a custom marshaling |
| 178 | + // This would require a custom struct and marshaling approach |
| 179 | + // For simplicity, we'll just add a note about it |
| 180 | + fmt.Fprintf(output, "# Object %d: Human-readable timestamps for reference:\n", decoder.count) |
| 181 | + for j, humanTime := range humanTimes { |
| 182 | + fmt.Fprintf(output, "# Sample %d: %s\n", j, humanTime) |
| 183 | + } |
| 184 | + } |
| 185 | + } |
| 186 | + |
| 187 | + // Output the JSON |
| 188 | + var jsonData []byte |
| 189 | + if *prettyPrint { |
| 190 | + jsonData, err = json.MarshalIndent(jsonStruct, "", " ") |
| 191 | + } else { |
| 192 | + jsonData, err = json.Marshal(jsonStruct) |
| 193 | + } |
| 194 | + |
| 195 | + if err != nil { |
| 196 | + fmt.Printf("Error encoding JSON object #%d to JSON: %v\n", decoder.count, err) |
| 197 | + continue |
| 198 | + } |
| 199 | + |
| 200 | + fmt.Fprintf(output, "# Object %d\n", decoder.count) |
| 201 | + fmt.Fprintln(output, string(jsonData)) |
| 202 | + } |
| 203 | + |
| 204 | + fmt.Printf("Successfully processed %d JSON objects\n", decoder.count) |
| 205 | +} |
0 commit comments