collector.go

Documentation: go.opencensus.io/stats/view

     1  // Copyright 2017, OpenCensus Authors
     2  //
     3  // Licensed under the Apache License, Version 2.0 (the "License");
     4  // you may not use this file except in compliance with the License.
     5  // You may obtain a copy of the License at
     6  //
     7  //     http://www.apache.org/licenses/LICENSE-2.0
     8  //
     9  // Unless required by applicable law or agreed to in writing, software
    10  // distributed under the License is distributed on an "AS IS" BASIS,
    11  // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    12  // See the License for the specific language governing permissions and
    13  // limitations under the License.
    14  //
    15  
    16  package view
    17  
    18  import (
    19  	"sort"
    20  	"time"
    21  
    22  	"go.opencensus.io/internal/tagencoding"
    23  	"go.opencensus.io/tag"
    24  )
    25  
    26  type collector struct {
    27  	// signatures holds the aggregations values for each unique tag signature
    28  	// (values for all keys) to its aggregator.
    29  	signatures map[string]AggregationData
    30  	// Aggregation is the description of the aggregation to perform for this
    31  	// view.
    32  	a *Aggregation
    33  }
    34  
    35  func (c *collector) addSample(s string, v float64, attachments map[string]interface{}, t time.Time) {
    36  	aggregator, ok := c.signatures[s]
    37  	if !ok {
    38  		aggregator = c.a.newData(t)
    39  		c.signatures[s] = aggregator
    40  	}
    41  	aggregator.addSample(v, attachments, t)
    42  }
    43  
    44  // collectRows returns a snapshot of the collected Row values.
    45  func (c *collector) collectedRows(keys []tag.Key) []*Row {
    46  	rows := make([]*Row, 0, len(c.signatures))
    47  	for sig, aggregator := range c.signatures {
    48  		tags := decodeTags([]byte(sig), keys)
    49  		row := &Row{Tags: tags, Data: aggregator.clone()}
    50  		rows = append(rows, row)
    51  	}
    52  	return rows
    53  }
    54  
    55  func (c *collector) clearRows() {
    56  	c.signatures = make(map[string]AggregationData)
    57  }
    58  
    59  // encodeWithKeys encodes the map by using values
    60  // only associated with the keys provided.
    61  func encodeWithKeys(m *tag.Map, keys []tag.Key) []byte {
    62  	// Compute the buffer length we will need ahead of time to avoid resizing later
    63  	reqLen := 0
    64  	for _, k := range keys {
    65  		s, _ := m.Value(k)
    66  		// We will store each key + its length
    67  		reqLen += len(s) + 1
    68  	}
    69  	vb := &tagencoding.Values{
    70  		Buffer: make([]byte, reqLen),
    71  	}
    72  	for _, k := range keys {
    73  		v, _ := m.Value(k)
    74  		vb.WriteValue([]byte(v))
    75  	}
    76  	return vb.Bytes()
    77  }
    78  
    79  // decodeTags decodes tags from the buffer and
    80  // orders them by the keys.
    81  func decodeTags(buf []byte, keys []tag.Key) []tag.Tag {
    82  	vb := &tagencoding.Values{Buffer: buf}
    83  	var tags []tag.Tag
    84  	for _, k := range keys {
    85  		v := vb.ReadValue()
    86  		if v != nil {
    87  			tags = append(tags, tag.Tag{Key: k, Value: string(v)})
    88  		}
    89  	}
    90  	vb.ReadIndex = 0
    91  	sort.Slice(tags, func(i, j int) bool { return tags[i].Key.Name() < tags[j].Key.Name() })
    92  	return tags
    93  }
    94
View as plain text