main.go

// Copyright (c) 2017 Uber Technologies, Inc.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
// THE SOFTWARE.

// This tool can be used to verify that the index files for multiple hosts that
// share the same shards are the same. If they're not identical, it will output
// the name of any series that are missing, or have mismatched checksums. The tool
// expects a directory where each subdirectory is the name of the host, and within
// each of those subdirectories is the "data" directory for that host, exactly as
// generated by M3DB itself. See the README for more details.

package main

import (
	"flag"
	"io"
	"io/ioutil"
	"log"
	_ "net/http/pprof"
	"os"
	"path"
	"strconv"
	"strings"
	"time"

	"github.com/m3db/m3/src/cmd/tools"
	"github.com/m3db/m3/src/dbnode/persist/fs"
	"github.com/m3db/m3x/ident"
	"github.com/m3db/m3x/pool"
)

var flagParser = flag.NewFlagSet("Verify Index files", flag.ExitOnError)

type seriesChecksums struct {
	host   string
	shard  uint32
	block  int64
	series seriesMap
}

type seriesMap map[string]series

type checksum uint32

type series struct {
	name     string
	checksum checksum
}

var (
	pathPrefixArg       = flagParser.String("path-prefix", "/var/lib/m3db", "Path prefix - must contain a folder called 'commitlogs'")
	namespaceArg        = flagParser.String("namespace", "metrics", "Namespace")
	shardsArg           = flagParser.String("shards", "", "Shards - set comma separated list of shards")
	blocksArgs          = flagParser.String("blocks", "", "Start unix timestamp (Seconds) - set comma separated list of unix timestamps")
	compareChecksumsArg = flagParser.Bool("compare-checksums", true, "Compare checksums")
)

var bytesPool pool.CheckedBytesPool

func main() {
	flagParser.Parse(os.Args[1:])

	log.SetOutput(os.Stdout)
	log.Println("initializing bytes pool...")
	bytesPool = tools.NewCheckedBytesPool()

	var (
		pathPrefix       = *pathPrefixArg
		namespaceStr     = *namespaceArg
		shardsVal        = *shardsArg
		blocksVal        = *blocksArgs
		compareChecksums = *compareChecksumsArg
	)

	blocks := parseBlockArgs(blocksVal)
	hosts, err := ioutil.ReadDir(pathPrefix)
	if err != nil {
		log.Fatalf("err reading dir: %s, err: %s\n", pathPrefix, err)
	}

	shards := parseShards(shardsVal)
	for _, block := range blocks {
		for _, shard := range shards {
			log.Printf("running test for shard: %d\n", shard)
			allHostSeriesChecksumsForShard := []seriesChecksums{}
			// Accumulate all the series checksums for each host for this shard
			for _, host := range hosts {
				hostShardReader, err := newReader(namespaceStr, pathPrefix, host.Name(), shard, time.Unix(block, 0))
				if err != nil {
					// Ignore folders for hosts that don't have this data
					if err == fs.ErrCheckpointFileNotFound {
						continue
					}
					log.Fatalf("err creating new reader: %s\n", err.Error())
				}
				hostShardSeriesChecksums := seriesChecksumsFromReader(hostShardReader, host.Name(), shard, block)
				allHostSeriesChecksumsForShard = append(allHostSeriesChecksumsForShard, hostShardSeriesChecksums)
			}

			allHostSeriesMapsForShard := []seriesMap{}
			for _, seriesChecksum := range allHostSeriesChecksumsForShard {
				allHostSeriesMapsForShard = append(allHostSeriesMapsForShard, seriesChecksum.series)
			}
			allSeriesForShard := mergeMaps(allHostSeriesMapsForShard...)
			for _, hostSeriesForShard := range allHostSeriesChecksumsForShard {
				compareSeriesChecksums(allSeriesForShard, hostSeriesForShard, compareChecksums)
			}
		}
	}
}

func seriesChecksumsFromReader(reader fs.DataFileSetReader, host string, shard uint32, block int64) seriesChecksums {
	seriesMap := seriesMap{}
	seriesChecksums := seriesChecksums{
		host:   host,
		series: seriesMap,
		shard:  shard,
		block:  block,
	}
	for {
		id, _, _, checksumVal, err := reader.ReadMetadata()
		if err == io.EOF {
			return seriesChecksums
		}
		if err != nil {
			log.Fatal("err reading from reader: ", err.Error())
		}
		idString := id.String()
		seriesMap[idString] = series{
			name:     idString,
			checksum: checksum(checksumVal),
		}
	}
}

func compareSeriesChecksums(against seriesMap, evaluate seriesChecksums, compareChecksums bool) {
	againstMap := against
	evaluateMap := evaluate.series
	missingSeries := []series{}
	checksumMismatchSeries := []series{}

	if len(againstMap) == len(evaluateMap) {
		log.Printf(
			"host %s has all %d series for shard: %d and block: %d",
			evaluate.host, len(againstMap), evaluate.shard, evaluate.block,
		)
	} else {
		log.Printf(
			"host %s has %d series, but there are a total of %d series in shard: %d and block %d\n",
			evaluate.host, len(evaluateMap), len(againstMap), evaluate.shard, evaluate.block,
		)
	}

	for seriesName, againstSeries := range againstMap {
		evaluateSeries, ok := evaluateMap[seriesName]

		if !ok {
			missingSeries = append(missingSeries, againstSeries)
			continue
		}

		if compareChecksums && againstSeries.checksum != evaluateSeries.checksum {
			checksumMismatchSeries = append(checksumMismatchSeries, againstSeries)
		}
	}

	for _, missing := range missingSeries {
		log.Printf("host %s is missing %s\n", evaluate.host, missing.name)
	}

	for _, mismatch := range checksumMismatchSeries {
		log.Printf("host %s has mismatching checksum for %s\n", evaluate.host, mismatch.name)
	}
}

func mergeMaps(seriesMaps ...seriesMap) seriesMap {
	merged := seriesMap{}
	for _, sMap := range seriesMaps {
		for key, val := range sMap {
			merged[key] = val
		}
	}
	return merged
}

func newReader(namespace, pathPrefix, hostName string, shard uint32, start time.Time) (fs.DataFileSetReader, error) {
	fsOpts := fs.NewOptions().SetFilePathPrefix(path.Join(pathPrefix, hostName))
	reader, err := fs.NewReader(bytesPool, fsOpts)
	if err != nil {
		return nil, err
	}

	openOpts := fs.DataReaderOpenOptions{
		Identifier: fs.FileSetFileIdentifier{
			Namespace:  ident.StringID(namespace),
			Shard:      shard,
			BlockStart: start,
		},
	}
	err = reader.Open(openOpts)
	return reader, err
}

func parseShards(shards string) []uint32 {
	var allShards []uint32

	if strings.TrimSpace(shards) == "" {
		return []uint32{}
	}

	// Handle commda-delimited shard list 1,3,5, etc
	for _, shard := range strings.Split(shards, ",") {
		shard = strings.TrimSpace(shard)
		if shard == "" {
			log.Fatalf("invalid shard list: '%s'", shards)
		}
		value, err := strconv.Atoi(shard)
		if err != nil {
			log.Fatalf("could not parse shard '%s': %v", shard, err)
		}
		allShards = append(allShards, uint32(value))
	}

	return allShards
}

func parseBlockArgs(blocks string) []int64 {
	allBlocks := []int64{}

	for _, block := range strings.Split(blocks, ",") {
		block = strings.TrimSpace(block)
		if block == "" {
			log.Fatalf("invalid block list: '%s'", block)
		}
		value, err := strconv.Atoi(block)
		if err != nil {
			log.Fatalf("could not parse blocks '%s': %v", block, err)
		}
		allBlocks = append(allBlocks, int64(value))
	}

	return allBlocks
}