// MarshalSplitCompress uses the stream compressor to marshal and compress sketch series payloads.
// If a compressed payload is larger than the max, a new payload will be generated. This method returns a slice of
// compressed protobuf marshaled gogen.SketchPayload objects. gogen.SketchPayload is not directly marshaled - instead
// it's contents are marshaled individually, packed with the appropriate protobuf metadata, and compressed in stream.
// The resulting payloads (when decompressed) are binary equal to the result of marshaling the whole object at once.
MarshalSplitCompress uses the stream compressor to marshal and compress sketch series payloads.
If a compressed payload is larger than the max, a new payload will be generated. This method returns a slice of
compressed protobuf marshaled gogen.SketchPayload objects. gogen.SketchPayload is not directly marshaled - instead
it's contents are marshaled individually, packed with the appropriate protobuf metadata, and compressed in stream.
The resulting payloads (when decompressed) are binary equal to the result of marshaling the whole object at once.
// MarshalSplitCompress uses the stream compressor to marshal and compress sketch series payloads . // If a compressed payload is larger than the max a new payload will be generated . This method returns a slice of // compressed protobuf marshaled gogen . SketchPayload objects . gogen . SketchPayload is not directly marshaled - instead // it ' s contents are marshaled individually packed with the appropriate protobuf metadata and compressed in stream . // The resulting payloads ( when decompressed ) are binary equal to the result of marshaling the whole object at once .
func (sl SketchSeriesList) MarshalSplitCompress(bufferContext *marshaler.BufferContext) ([]*[]byte, error) {
	footer := []byte{0x12, 0}
	bufferContext.CompressorInput.Reset()
	bufferContext.CompressorOutput.Reset()
	compressor, e := stream.NewCompressor(bufferContext.CompressorInput, bufferContext.CompressorOutput, []byte{}, footer, []byte{})
	if e != nil {
		return nil, e
	}
	payloads := []*[]byte{}
	dsl := make([]gogen.SketchPayload_Sketch_Dogsketch, 1)
	for _, ss := range sl {
		if len(ss.Points) > cap(dsl) {
			dsl = make([]gogen.SketchPayload_Sketch_Dogsketch, len(ss.Points))
		}
		for i, p := range ss.Points {
			b := p.Sketch.Basic
			k, n := p.Sketch.Cols()
			dsl[i] = gogen.SketchPayload_Sketch_Dogsketch{
				Ts:  p.Ts,
				Cnt: b.Cnt,
				Min: b.Min,
				Max: b.Max,
				Avg: b.Avg,
				Sum: b.Sum,
				K:   k,
				N:   n,
			}
		}
		sketch := gogen.SketchPayload_Sketch{
			Metric:      ss.Name,
			Host:        ss.Host,
			Tags:        ss.Tags,
			Dogsketches: dsl[:len(ss.Points)],
		}
		metadataSize := 0
		bufferContext.PrecompressionBuf[metadataSize] = 0xa
		metadataSize++
		metadataSize = encodeVarintAgentPayload(bufferContext.PrecompressionBuf, metadataSize, uint64(sketch.Size()))
		totalItemSize := sketch.Size() + metadataSize
		if totalItemSize > cap(bufferContext.PrecompressionBuf) {
			bufferContext.PrecompressionBuf = append(bufferContext.PrecompressionBuf, make([]byte, totalItemSize-cap(bufferContext.PrecompressionBuf))...)
			bufferContext.PrecompressionBuf = bufferContext.PrecompressionBuf[:cap(bufferContext.PrecompressionBuf)]
		}
		_, e := sketch.MarshalTo(bufferContext.PrecompressionBuf[metadataSize:])
		if e != nil {
			return nil, e
		}
		switch compressor.AddItem(bufferContext.PrecompressionBuf[:totalItemSize]) {
		case stream.ErrPayloadFull:
			expvarsPayloadFull.Add(1)
			tlmPayloadFull.Inc()
			payload, e := compressor.Close()
			if e != nil {
				return nil, e
			}
			payloads = append(payloads, &payload)
			bufferContext.CompressorInput.Reset()
			bufferContext.CompressorOutput.Reset()
			compressor, e = stream.NewCompressor(bufferContext.CompressorInput, bufferContext.CompressorOutput, []byte{}, footer, []byte{})
			if e != nil {
				return nil, e
			}
			e = compressor.AddItem(bufferContext.PrecompressionBuf[:totalItemSize])
			if e == stream.ErrItemTooBig {
				expvarsItemTooBig.Add(1)
				tlmItemTooBig.Inc()
				continue
			}
			if e != nil {
				expvarsUnexpectedItemDrops.Add(1)
				tlmUnexpectedItemDrops.Inc()
				return nil, e
			}
		case stream.ErrItemTooBig:
			expvarsItemTooBig.Add(1)
			tlmItemTooBig.Add(1)
		case nil:
			continue
		default:
			expvarsUnexpectedItemDrops.Add(1)
			tlmUnexpectedItemDrops.Inc()
			return nil, e
		}
	}
	payload, e := compressor.Close()
	if e != nil {
		return nil, e
	}
	payloads = append(payloads, &payload)
	return payloads, nil
}
func ( sl SketchSeriesList ) MarshalSplitCompress ( bufferContext * marshaler . BufferContext ) ( [ ] * [ ] byte , error ) { footer := [ ] byte { 0x12 , 0 } bufferContext . CompressorInput . Reset ( ) bufferContext . CompressorOutput . Reset ( ) compressor , e := stream . NewCompressor ( bufferContext . CompressorInput , bufferContext . CompressorOutput , [ ] byte { } , footer , [ ] byte { } ) if e != nil { return nil , e } payloads := [ ] * [ ] byte { } dsl := make ( [ ] gogen . SketchPayload_Sketch_Dogsketch , 1 ) for _ , ss := range sl { if len ( ss . Points ) > cap ( dsl ) { dsl = make ( [ ] gogen . SketchPayload_Sketch_Dogsketch , len ( ss . Points ) ) } for i , p := range ss . Points { b := p . Sketch . Basic k , n := p . Sketch . Cols ( ) dsl [ i ] = gogen . SketchPayload_Sketch_Dogsketch { Ts : p . Ts , Cnt : b . Cnt , Min : b . Min , Max : b . Max , Avg : b . Avg , Sum : b . Sum , K : k , N : n , } } sketch := gogen . SketchPayload_Sketch { Metric : ss . Name , Host : ss . Host , Tags : ss . Tags , Dogsketches : dsl [ : len ( ss . Points ) ] , } metadataSize := 0 bufferContext . PrecompressionBuf [ metadataSize ] = 0xa metadataSize ++ metadataSize = encodeVarintAgentPayload ( bufferContext . PrecompressionBuf , metadataSize , uint64 ( sketch . Size ( ) ) ) totalItemSize := sketch . Size ( ) + metadataSize if totalItemSize > cap ( bufferContext . PrecompressionBuf ) { bufferContext . PrecompressionBuf = append ( bufferContext . PrecompressionBuf , make ( [ ] byte , totalItemSize - cap ( bufferContext . PrecompressionBuf ) ) ... ) bufferContext . PrecompressionBuf = bufferContext . PrecompressionBuf [ : cap ( bufferContext . PrecompressionBuf ) ] } _ , e := sketch . MarshalTo ( bufferContext . PrecompressionBuf [ metadataSize : ] ) if e != nil { return nil , e } switch compressor . AddItem ( bufferContext . PrecompressionBuf [ : totalItemSize ] ) { case stream . ErrPayloadFull : expvarsPayloadFull . Add ( 1 ) tlmPayloadFull . Inc ( ) payload , e := compressor . Close ( ) if e != nil { return nil , e } payloads = append ( payloads , & payload ) bufferContext . CompressorInput . Reset ( ) bufferContext . CompressorOutput . Reset ( ) compressor , e = stream . NewCompressor ( bufferContext . CompressorInput , bufferContext . CompressorOutput , [ ] byte { } , footer , [ ] byte { } ) if e != nil { return nil , e } e = compressor . AddItem ( bufferContext . PrecompressionBuf [ : totalItemSize ] ) if e == stream . ErrItemTooBig { expvarsItemTooBig . Add ( 1 ) tlmItemTooBig . Inc ( ) continue } if e != nil { expvarsUnexpectedItemDrops . Add ( 1 ) tlmUnexpectedItemDrops . Inc ( ) return nil , e } case stream . ErrItemTooBig : expvarsItemTooBig . Add ( 1 ) tlmItemTooBig . Add ( 1 ) case nil : continue default : expvarsUnexpectedItemDrops . Add ( 1 ) tlmUnexpectedItemDrops . Inc ( ) return nil , e } } payload , e := compressor . Close ( ) if e != nil { return nil , e } payloads = append ( payloads , & payload ) return payloads , nil }
MarshalSplitCompress uses the stream compressor to marshal and compress sketch series payloads.
MarshalSplitCompress uses the stream compressor to marshal and compress sketch series payloads .
// The Metadata field of gogen.SketchPayload is never written to - so pack an empty metadata as the footer // Pack the protobuf metadata - see SketchPayload.MarshalTo in agent_payload.pb.go for reference. // Magic number that occurs before the varint encoding // Resize the pre-compression buffer if needed // Marshal the sketch to the precompression buffer after the metadata // Compress the protobuf metadata and the marshaled sketch // Since the compression buffer is full - flush it and rotate // Add it to the new compression buffer // Item was too big, drop it // Unexpected error bail out // Item was too big, drop it // Unexpected error bail out // return payloads, nonCompressed