// Chunk returns the chunk for the reference number.
178124
func (h *headChunkReader) Chunk(ref uint64) (chunkenc.Chunk, error) {
	sid, cid := unpackChunkID(ref)

	s := h.head.series.getByID(sid)
	// This means that the series has been garbage collected.
	if s == nil {
		return nil, ErrNotFound
	}

	s.Lock()
	c := s.chunk(int(cid))

	// This means that the chunk has been garbage collected or is outside
	// the specified range.
	if c == nil || !c.OverlapsClosedInterval(h.mint, h.maxt) {
		s.Unlock()
		return nil, ErrNotFound
	}
	s.Unlock()

	return &safeChunk{
		Chunk: c.chunk,
		s:     s,
		cid:   int(cid),
	}, nil
}
659170
func (c *Chunk) CopyConstruct() *Chunk {
	newChk := &Chunk{numVirtualRows: c.numVirtualRows, capacity: c.capacity, columns: make([]*column, len(c.columns))}
	for i := range c.columns {
		newChk.columns[i] = c.columns[i].copyConstruct()
	}
	return newChk
}
99390
func (ef *File) FetchRawChunk(offset int64) (Chunk, error) {
	ef.Lock()
	defer ef.Unlock()
	c := NewChunk()
	GoToSeeker(ef.file, offset)
	c.Offset = offset
	c.Data = make([]byte, ChunkHeaderSize)
	if _, err := ef.file.Read(c.Data); err != nil {
		return c, err
	}
	reader := bytes.NewReader(c.Data)
	c.ParseChunkHeader(reader)
	return c, nil
}
1024803
func (h *hasherStore) Put(ctx context.Context, chunkData ChunkData) (Reference, error) {
	c := chunkData
	var encryptionKey encryption.Key
	if h.toEncrypt {
		var err error
		c, encryptionKey, err = h.encryptChunkData(chunkData)
		if err != nil {
			return nil, err
		}
	}
	chunk := h.createChunk(c)
	h.storeChunk(ctx, chunk)

	return Reference(append(chunk.Address(), encryptionKey...)), nil
}
525485
func (u *multiuploader) readChunk(ch chan chunk) {
	defer u.wg.Done()
	for {
		data, ok := <-ch

		if !ok {
			break
		}

		if u.geterr() == nil {
			if err := u.send(data); err != nil {
				u.seterr(err)
			}
		}
	}
}
353792
func (pb *Block) Chunks() (ChunkReader, error) {
	if err := pb.startRead(); err != nil {
		return nil, err
	}
	return blockChunkReader{ChunkReader: pb.chunkr, b: pb}, nil
}
40246
func (ch *Chunk) Jump(bytesAhead int) error {
	var err error
	var n int64
	if bytesAhead > 0 {
		n, err = io.CopyN(ioutil.Discard, ch.R, int64(bytesAhead))
		ch.Pos += int(n)
	}
	return err
}
1244905
func EventFromChunk(sessionID string, chunk *SessionChunk) (EventFields, error) {
	var fields EventFields
	eventStart := time.Unix(0, chunk.Time).In(time.UTC).Round(time.Millisecond)
	err := json.Unmarshal(chunk.Data, &fields)
	if err != nil {
		return nil, trace.Wrap(err)
	}
	fields[SessionEventID] = sessionID
	fields[EventIndex] = chunk.EventIndex
	fields[EventTime] = eventStart
	fields[EventType] = chunk.EventType
	if fields[EventID] == "" {
		fields[EventID] = uuid.New()
	}
	return fields, nil
}
832221
func numChunks(b []byte) int {
	lenB := len(b)
	if lenB <= ChunkSize {
		return 1
	}
	return len(b)/chunkedDataLen + 1
}
206565
func readChunk(file *os.File, offset int64, buf []byte) (err error) {
	n, err := file.ReadAt(buf, offset)
	if err != nil {
		return
	}
	if n != ChunkSize {
		return fmt.Errorf("Invalid read %v", n)
	}
	return
}
826010
func (p *Pool) PutChunk(fields []*types.FieldType, chk *Chunk) {
	for i, f := range fields {
		switch elemLen := getFixedLen(f); elemLen {
		case varElemLen:
			p.varLenColPool.Put(chk.columns[i])
		case 4:
			p.fixLenColPool4.Put(chk.columns[i])
		case 8:
			p.fixLenColPool8.Put(chk.columns[i])
		case 16:
			p.fixLenColPool16.Put(chk.columns[i])
		case 40:
			p.fixLenColPool40.Put(chk.columns[i])
		}
	}
	chk.columns = nil // release the column references.
}
77420
func (cb *CountBytes) Bytes() int64 {
	cb.mu.Lock()
	defer cb.mu.Unlock()
	return cb.bytes
}
579145
func (s *CatIndicesService) Bytes(bytes string) *CatIndicesService {
	s.bytes = bytes
	return s
}
1252022
func (i Image) Reference() string {
	if len(i.Digest.String()) > 1 {
		return i.Digest.String()
	}

	return i.Tag
}
125912
func (f *asyncFlusher) upload(ctx context.Context, chunk chunk) (int, error) {
	// Give up right away if the context is already dead.
	if err := ctx.Err(); err != nil {
		logging.WithError(err).Errorf(ctx, "Skipping upload")
		return 0, err
	}

	// Collect all pending data into an array of rows. We cheat here when
	// unpacking gob-serialized entries and use 'bigquery.JsonValue' instead of
	// 'biquery.Value{}' in Data. They are compatible. This cheat avoids to do
	// a lot of allocations just to appease the type checker.
	var rows []*bqapi.TableDataInsertAllRequestRows
	var entries []rawEntry

	for _, task := range chunk.Tasks {
		ctx := logging.SetField(ctx, "name", task.Name)

		if err := gob.NewDecoder(bytes.NewReader(task.Payload)).Decode(&entries); err != nil {
			logging.WithError(err).Errorf(ctx, "Failed to gob-decode pending batch, it will be skipped")
			continue
		}

		for i, entry := range entries {
			insertID := entry.InsertID
			if insertID == "" {
				// The task names are autogenerated and guaranteed to be unique.
				// Use them as a base for autogenerated insertID.
				insertID = fmt.Sprintf("bqlog:%s:%d", task.Name, i)
			}
			rows = append(rows, &bqapi.TableDataInsertAllRequestRows{
				InsertId: insertID,
				Json:     entry.Data,
			})
			// We need to nil Data maps here to be able to reuse 'entries' array
			// capacity later. Otherwise gob decode "discovers" maps and overwrites
			// their data in-place, spoiling 'rows'.
			entries[i].Data = nil
		}

		entries = entries[:0]
	}

	if len(rows) == 0 {
		chunk.Done(ctx)
		return 0, nil
	}

	// Now actually send all the entries with retries.
	var lastResp *bqapi.TableDataInsertAllResponse
	taggedCtx := clock.Tag(ctx, "insert-retry") // used by tests
	err := retry.Retry(taggedCtx, transient.Only(f.retryParams), func() error {
		startTime := clock.Now(ctx)
		var err error
		lastResp, err = f.Insert(ctx, &bqapi.TableDataInsertAllRequest{
			SkipInvalidRows: true, // they will be reported in lastResp.InsertErrors
			Rows:            rows,
		})
		code := 0
		status := "ok"
		if gerr, _ := err.(*googleapi.Error); gerr != nil {
			code = gerr.Code
			status = fmt.Sprintf("http_%d", code)
		} else if ctx.Err() != nil {
			status = "timeout"
		} else if err != nil {
			status = "unknown"
		}
		dt := clock.Since(ctx, startTime)
		bigQueryLatency.Add(ctx, float64(dt.Nanoseconds()/1e6), f.TableRef, "insertAll", status)
		if code >= 500 {
			return transient.Tag.Apply(err)
		}
		return err
	}, func(err error, wait time.Duration) {
		logging.Fields{
			logging.ErrorKey: err,
			"wait":           wait,
		}.Warningf(ctx, "Failed to send data to BigQuery")
	})
	if err != nil {
		logging.WithError(err).Errorf(ctx, "Failed to send data to BigQuery")
		if !transient.Tag.In(err) && err != context.DeadlineExceeded {
			chunk.Done(ctx)
		}
		return 0, err
	}

	if success := len(rows) - len(lastResp.InsertErrors); success > 0 {
		flushedEntryCount.Add(ctx, int64(success), f.TableRef, "ok")
	}

	if len(lastResp.InsertErrors) != 0 {
		// Use only first error as a sample. Dumping them all is impractical.
		blob, _ := json.MarshalIndent(lastResp.InsertErrors[0].Errors, "", "  ")
		logging.Errorf(ctx, "%d rows weren't accepted, sample error:\n%s", len(lastResp.InsertErrors), blob)

		// Categorize errors by reason to dump them to monitoring. We look only
		// at first suberror.
		perReason := make(map[string]int64, 5)
		for _, err := range lastResp.InsertErrors {
			reason := "unknown"
			if len(err.Errors) > 0 {
				reason = err.Errors[0].Reason // usually just "invalid"
			}
			perReason[reason]++
		}
		for reason, count := range perReason {
			flushedEntryCount.Add(ctx, count, f.TableRef, reason)
		}
	}

	chunk.Done(ctx)
	return len(rows), nil
}
1043000
func (l chunkByIndex) Less(i, j int) bool {
	ii := l[i].Index()
	ij := l[j].Index()
	return ii < ij
}
1154892
func (a *AggMetric) add(ts uint32, val float64) {
	t0 := ts - (ts % a.chunkSpan)

	if len(a.chunks) == 0 {
		chunkCreate.Inc()
		// no data has been added to this AggMetric yet.
		// note that we may not be aware of prior data that belongs into this chunk
		// so we should track this cutoff point
		a.chunks = append(a.chunks, chunk.NewFirst(t0))
		a.firstTs = ts

		if err := a.chunks[0].Push(ts, val); err != nil {
			panic(fmt.Sprintf("FATAL ERROR: this should never happen. Pushing initial value <%d,%f> to new chunk at pos 0 failed: %q", ts, val, err))
		}
		totalPoints.Inc()

		log.Debugf("AM: %s Add(): created first chunk with first point: %v", a.key, a.chunks[0])
		a.lastWrite = uint32(time.Now().Unix())
		if a.dropFirstChunk {
			a.lastSaveStart = t0
			a.lastSaveFinish = t0
		}
		a.addAggregators(ts, val)
		return
	}

	currentChunk := a.chunks[a.currentChunkPos]

	if t0 == currentChunk.Series.T0 {
		// last prior data was in same chunk as new point
		if currentChunk.Series.Finished {
			// if we've already 'finished' the chunk, it means it has the end-of-stream marker and any new points behind it wouldn't be read by an iterator
			// you should monitor this metric closely, it indicates that maybe your GC settings don't match how you actually send data (too late)
			discardedReceivedTooLate.Inc()
			PromDiscardedSamples.WithLabelValues(receivedTooLate, strconv.Itoa(int(a.key.MKey.Org))).Inc()
			return
		}

		if err := currentChunk.Push(ts, val); err != nil {
			log.Debugf("AM: failed to add metric to chunk for %s. %s", a.key, err)
			a.discardedMetricsInc(err)
			return
		}
		totalPoints.Inc()
		a.lastWrite = uint32(time.Now().Unix())

		if log.IsLevelEnabled(log.DebugLevel) {
			log.Debugf("AM: %s Add(): pushed new value to last chunk: %v", a.key, a.chunks[0])
		}
	} else if t0 < currentChunk.Series.T0 {
		log.Debugf("AM: Point at %d has t0 %d, goes back into previous chunk. CurrentChunk t0: %d, LastTs: %d", ts, t0, currentChunk.Series.T0, currentChunk.Series.T)
		discardedSampleOutOfOrder.Inc()
		PromDiscardedSamples.WithLabelValues(sampleOutOfOrder, strconv.Itoa(int(a.key.MKey.Org))).Inc()
		return
	} else {
		// Data belongs in a new chunk.

		// If it isn't finished already, add the end-of-stream marker and flag the chunk as "closed"
		currentChunk.Finish()

		a.pushToCache(currentChunk)
		// If we are a primary node, then add the chunk to the write queue to be saved to Cassandra
		if cluster.Manager.IsPrimary() {
			log.Debugf("AM: persist(): node is primary, saving chunk. %s T0: %d", a.key, currentChunk.Series.T0)
			// persist the chunk. If the writeQueue is full, then this will block.
			a.persist(a.currentChunkPos)
		}

		a.currentChunkPos++
		if a.currentChunkPos >= int(a.numChunks) {
			a.currentChunkPos = 0
		}

		chunkCreate.Inc()
		if len(a.chunks) < int(a.numChunks) {
			a.chunks = append(a.chunks, chunk.New(t0))
			if err := a.chunks[a.currentChunkPos].Push(ts, val); err != nil {
				panic(fmt.Sprintf("FATAL ERROR: this should never happen. Pushing initial value <%d,%f> to new chunk at pos %d failed: %q", ts, val, a.currentChunkPos, err))
			}
			totalPoints.Inc()
			log.Debugf("AM: %s Add(): added new chunk to buffer. now %d chunks. and added the new point: %s", a.key, a.currentChunkPos+1, a.chunks[a.currentChunkPos])
		} else {
			chunkClear.Inc()
			totalPoints.DecUint64(uint64(a.chunks[a.currentChunkPos].NumPoints))

			a.chunks[a.currentChunkPos] = chunk.New(t0)
			if err := a.chunks[a.currentChunkPos].Push(ts, val); err != nil {
				panic(fmt.Sprintf("FATAL ERROR: this should never happen. Pushing initial value <%d,%f> to new chunk at pos %d failed: %q", ts, val, a.currentChunkPos, err))
			}
			totalPoints.Inc()
			log.Debugf("AM: %s Add(): cleared chunk at %d of %d and replaced with new. and added the new point: %s", a.key, a.currentChunkPos, len(a.chunks), a.chunks[a.currentChunkPos])
		}
		a.lastWrite = uint32(time.Now().Unix())

	}
	a.addAggregators(ts, val)
}