diff --git a/storage/local/persistence.go b/storage/local/persistence.go index da4ebfde2a..9e0859224e 100644 --- a/storage/local/persistence.go +++ b/storage/local/persistence.go @@ -349,7 +349,26 @@ func (p *persistence) recoverFromCrash(fingerprintToSeries map[clientmodel.Finge return nil } -// TODO: Document. +// sanitizeSeries sanitizes a series based on its series file as defined by the provided directory and FileInfo. +// The method returns the fingerprint as derived from the directory and file name, and whether the provided +// file has been sanatizide. A file that failed to be sanitized is deleted, if possible. +// +// The following steps are performed: +// +// - A file whose name doesn't comply with the naming scheme of a series file is simply deleted. +// +// - If the size of the series file isn't a multiple of the chunk size, extraneous bytes are truncated. +// If the truncation fails, the file is deleted instead. +// +// - A file that is empty (after truncation) is deleted. +// +// - A series that is not archived (i.e. it is in the fingerprintToSeries map) is checked for consistency of +// its various parameters (like head-chunk persistence state, offset of chunkDescs etc.). In particular, +// overlap between an in-memory head chunk with the most recent persisted chunk is checked. Inconsistencies +// are rectified. +// +// - A series this in archived (i.e. it is not in the fingerprintToSeries map) is checked for its presence +// in the index of archived series. If it cannot be found there, it is deleted. func (p *persistence) sanitizeSeries(dirname string, fi os.FileInfo, fingerprintToSeries map[clientmodel.Fingerprint]*memorySeries) (clientmodel.Fingerprint, bool) { filename := path.Join(dirname, fi.Name()) purge := func() { @@ -364,7 +383,11 @@ func (p *persistence) sanitizeSeries(dirname string, fi os.FileInfo, fingerprint purge() return fp, false } - fp.LoadFromString(path.Base(dirname) + fi.Name()[:fpLen-seriesDirNameLen]) // TODO: Panics if that doesn't parse as hex. + if err := fp.LoadFromString(path.Base(dirname) + fi.Name()[:fpLen-seriesDirNameLen]); err != nil { + glog.Warningf("Error parsing file name %s: %s", filename, err) + purge() + return fp, false + } bytesToTrim := fi.Size() % int64(p.chunkLen+chunkHeaderLen) chunksInFile := int(fi.Size()) / (p.chunkLen + chunkHeaderLen) @@ -676,9 +699,6 @@ func (p *persistence) persistChunk(fp clientmodel.Fingerprint, c chunk) (int, er // each index in indexes. It is the caller's responsibility to not persist or // drop anything for the same fingerprint concurrently. func (p *persistence) loadChunks(fp clientmodel.Fingerprint, indexes []int, indexOffset int) ([]chunk, error) { - // TODO: we need to verify at some point that file length is a multiple of - // the chunk size. When is the best time to do this, and where to remember - // it? Right now, we only do it when loading chunkDescs. f, err := p.openChunkFileForReading(fp) if err != nil { return nil, err @@ -731,15 +751,11 @@ func (p *persistence) loadChunkDescs(fp clientmodel.Fingerprint, beforeTime clie } totalChunkLen := chunkHeaderLen + p.chunkLen if fi.Size()%int64(totalChunkLen) != 0 { - // TODO: record number of encountered corrupt series files in a metric? - - // Truncate the file size to the nearest multiple of chunkLen. - truncateTo := fi.Size() - fi.Size()%int64(totalChunkLen) - glog.Infof("Bad series file size for %s: %d bytes (no multiple of %d). Truncating to %d bytes.", fp, fi.Size(), totalChunkLen, truncateTo) - // TODO: this doesn't work, as this is a read-only file handle. - if err := f.Truncate(truncateTo); err != nil { - return nil, err - } + p.setDirty(true) + return nil, fmt.Errorf( + "size of series file for fingerprint %v is %d, which is not a multiple of the chunk length %d", + fp, fi.Size(), totalChunkLen, + ) } numChunks := int(fi.Size()) / totalChunkLen @@ -773,6 +789,42 @@ func (p *persistence) loadChunkDescs(fp clientmodel.Fingerprint, beforeTime clie // checkpointSeriesMapAndHeads persists the fingerprint to memory-series mapping // and all open (non-full) head chunks. Do not call concurrently with // loadSeriesMapAndHeads. +// +// Description of the file format: +// +// (1) Magic string (const headsMagicString). +// +// (2) Varint-encoded format version (const headsFormatVersion). +// +// (3) Number of series in checkpoint as big-endian uint64. +// +// (4) Repeated once per series: +// +// (4.1) A flag byte, see flag constants above. +// +// (4.2) The fingerprint as big-endian uint64. +// +// (4.3) The metric as defined by codable.Metric. +// +// (4.4) The varint-encoded chunkDescsOffset. +// +// (4.5) The varint-encoded savedFirstTime. +// +// (4.6) The varint-encoded number of chunk descriptors. +// +// (4.7) Repeated once per chunk descriptor, oldest to most recent: +// +// (4.7.1) The varint-encoded first time. +// +// (4.7.2) The varint-encoded last time. +// +// (4.8) Exception to 4.7: If the most recent chunk is a non-persisted head chunk, +// the following is persisted instead of the most recent chunk descriptor: +// +// (4.8.1) A byte defining the chunk type. +// +// (4.8.2) The head chunk itself, marshaled with the marshal() method. +// func (p *persistence) checkpointSeriesMapAndHeads(fingerprintToSeries *seriesMap, fpLocker *fingerprintLocker) (err error) { glog.Info("Checkpointing in-memory metrics and head chunks...") begin := time.Now() @@ -1328,7 +1380,18 @@ func (p *persistence) openChunkFileForWriting(fp clientmodel.Fingerprint) (*os.F if err := os.MkdirAll(p.dirNameForFingerprint(fp), 0700); err != nil { return nil, err } - return os.OpenFile(p.fileNameForFingerprint(fp), os.O_WRONLY|os.O_APPEND|os.O_CREATE, 0640) + f, err := os.OpenFile(p.fileNameForFingerprint(fp), os.O_WRONLY|os.O_APPEND|os.O_CREATE, 0640) + if err != nil { + return f, err + } + offset, err := f.Seek(0, os.SEEK_CUR) + if offset%int64(chunkHeaderLen+p.chunkLen) != 0 { + return f, fmt.Errorf( + "size of series file for fingerprint %v is %d, which is not a multiple of the chunk length %d", + fp, offset, chunkHeaderLen+p.chunkLen, + ) + } + return f, err } func (p *persistence) openChunkFileForReading(fp clientmodel.Fingerprint) (*os.File, error) { diff --git a/storage/local/preload.go b/storage/local/preload.go index 1e1cd6c96b..ef5f110e07 100644 --- a/storage/local/preload.go +++ b/storage/local/preload.go @@ -99,10 +99,6 @@ func (p *memorySeriesPreloader) GetMetricRangeAtInterval(fp clientmodel.Fingerpr // Close implements Preloader. func (p *memorySeriesPreloader) Close() { - // TODO: Idea about a primitive but almost free heuristic to not evict - // "recently used" chunks: Do not unpin the chunks right here, but hand - // over the pinnedChunkDescs to a manager that will delay the unpinning - // based on time and memory pressure. for _, cd := range p.pinnedChunkDescs { cd.unpin(p.storage.evictRequests) } diff --git a/storage/local/series.go b/storage/local/series.go index 62dc84e019..9d2286067a 100644 --- a/storage/local/series.go +++ b/storage/local/series.go @@ -281,8 +281,7 @@ func (s *memorySeries) preloadChunks(indexes []int, mss *memorySeriesStorage) ([ panic("requested loading chunks from persistence in a situation where we must not have persisted data for chunk descriptors in memory") } fp := s.metric.Fingerprint() - // TODO: Remove law-of-Demeter violation? - chunks, err := mss.persistence.loadChunks(fp, loadIndexes, s.chunkDescsOffset) + chunks, err := mss.loadChunks(fp, loadIndexes, s.chunkDescsOffset) if err != nil { // Unpin the chunks since we won't return them as pinned chunks now. for _, cd := range pinnedChunkDescs { @@ -343,8 +342,7 @@ func (s *memorySeries) preloadChunksForRange( firstChunkDescTime = s.chunkDescs[0].firstTime() } if s.chunkDescsOffset != 0 && from.Before(firstChunkDescTime) { - // TODO: Remove law-of-demeter violation? - cds, err := mss.persistence.loadChunkDescs(fp, firstChunkDescTime) + cds, err := mss.loadChunkDescs(fp, firstChunkDescTime) if err != nil { return nil, err } diff --git a/storage/local/storage.go b/storage/local/storage.go index f19808eb0f..7544e0149a 100644 --- a/storage/local/storage.go +++ b/storage/local/storage.go @@ -774,6 +774,16 @@ func (s *memorySeriesStorage) purgeSeries(fp clientmodel.Fingerprint, beforeTime s.persistence.updateArchivedTimeRange(fp, newFirstTime, lastTime) } +// See persistence.loadChunks for detailed explanation. +func (s *memorySeriesStorage) loadChunks(fp clientmodel.Fingerprint, indexes []int, indexOffset int) ([]chunk, error) { + return s.persistence.loadChunks(fp, indexes, indexOffset) +} + +// See persistence.loadChunkDescs for detailed explanation. +func (s *memorySeriesStorage) loadChunkDescs(fp clientmodel.Fingerprint, beforeTime clientmodel.Timestamp) ([]*chunkDesc, error) { + return s.persistence.loadChunkDescs(fp, beforeTime) +} + // To expose persistQueueCap as metric: var ( persistQueueCapDesc = prometheus.NewDesc(