<commit_before>package bigtable

import (
	"context"
	"flag"
	"fmt"
	"sync"
	"time"

	"cloud.google.com/go/bigtable"
	"github.com/grafana/metrictank/cluster"
	"github.com/grafana/metrictank/idx"
	"github.com/grafana/metrictank/idx/memory"
	"github.com/grafana/metrictank/stats"
	"github.com/raintank/schema"
	"github.com/rakyll/globalconf"
	log "github.com/sirupsen/logrus"
)

const COLUMN_FAMILY = "idx"

var (
	// metric idx.bigtable.query-insert.ok is how many insert queries for a metric completed successfully (triggered by an add or an update)
	statQueryInsertOk = stats.NewCounter32("idx.bigtable.query-insert.ok")
	// metric idx.bigtable.query-insert.fail is how many insert queries for a metric failed (triggered by an add or an update)
	statQueryInsertFail = stats.NewCounter32("idx.bigtable.query-insert.fail")
	// metric idx.bigtable.query-delete.ok is how many delete queries for a metric completed successfully (triggered by an update or a delete)
	statQueryDeleteOk = stats.NewCounter32("idx.bigtable.query-delete.ok")
	// metric idx.bigtable.query-delete.fail is how many delete queries for a metric failed (triggered by an update or a delete)
	statQueryDeleteFail = stats.NewCounter32("idx.bigtable.query-delete.fail")

	// metric idx.bigtable.query-insert.wait is time inserts spent in queue before being executed
	statQueryInsertWaitDuration = stats.NewLatencyHistogram12h32("idx.bigtable.query-insert.wait")
	// metric idx.bigtable.query-insert.exec is time spent executing inserts (possibly repeatedly until success)
	statQueryInsertExecDuration = stats.NewLatencyHistogram15s32("idx.bigtable.query-insert.exec")
	// metric idx.bigtable.query-delete.exec is time spent executing deletes (possibly repeatedly until success)
	statQueryDeleteExecDuration = stats.NewLatencyHistogram15s32("idx.bigtable.query-delete.exec")

	// metric idx.bigtable.add is the duration of an add of one metric to the bigtable idx, including the add to the in-memory index, excluding the insert query
	statAddDuration = stats.NewLatencyHistogram15s32("idx.bigtable.add")
	// metric idx.bigtable.update is the duration of an update of one metric to the bigtable idx, including the update to the in-memory index, excluding any insert/delete queries
	statUpdateDuration = stats.NewLatencyHistogram15s32("idx.bigtable.update")
	// metric idx.bigtable.prune is the duration of a prune of the bigtable idx, including the prune of the in-memory index and all needed delete queries
	statPruneDuration = stats.NewLatencyHistogram15s32("idx.bigtable.prune")
	// metric idx.bigtable.delete is the duration of a delete of one or more metrics from the bigtable idx, including the delete from the in-memory index and the delete query
	statDeleteDuration = stats.NewLatencyHistogram15s32("idx.bigtable.delete")
	// metric idx.bigtable.save.skipped is how many saves have been skipped due to the writeQueue being full
	statSaveSkipped = stats.NewCounter32("idx.bigtable.save.skipped")
	// metric idx.bigtable.save.bytes-per-request is the number of bytes written to bigtable in each request.
	statSaveBytesPerRequest = stats.NewMeter32("idx.bigtable.save.bytes-per-request", true)

	maxStale          time.Duration
	pruneInterval     time.Duration
	updateIdx         bool
	updateInterval    time.Duration
	updateInterval32  uint32
	writeQueueSize    int
	writeConcurrency  int
	writeMaxFlushSize int

	Enabled          bool
	bigtableInstance string
	gcpProject       string
	tableName        string
	createCF         bool
)

func ConfigSetup() {
	btIdx := flag.NewFlagSet("bigtable-idx", flag.ExitOnError)

	btIdx.BoolVar(&Enabled, "enabled", true, "")
	btIdx.StringVar(&gcpProject, "gcp-project", "default", "Name of GCP project the bigtable cluster resides in")
	btIdx.StringVar(&bigtableInstance, "bigtable-instance", "default", "Name of bigtable instance")
	btIdx.StringVar(&tableName, "table-name", "metric_idx", "Name of bigtable table used for metricDefs")
	btIdx.IntVar(&writeQueueSize, "write-queue-size", 100000, "Max number of metricDefs allowed to be unwritten to bigtable. Must be larger then write-max-flush-size")
	btIdx.IntVar(&writeConcurrency, "write-concurrency", 5, "Number of writer threads to use")
	btIdx.IntVar(&writeMaxFlushSize, "write-max-flush-size", 10000, "Max number of metricDefs in each batch write to bigtable")
	btIdx.BoolVar(&updateIdx, "update-bigtable-index", true, "synchronize index changes to bigtable. not all your nodes need to do this.")
	btIdx.DurationVar(&updateInterval, "update-interval", time.Hour*3, "frequency at which we should update the metricDef lastUpdate field, use 0s for instant updates")
	btIdx.DurationVar(&maxStale, "max-stale", 0, "clear series from the index if they have not been seen for this much time.")
	btIdx.DurationVar(&pruneInterval, "prune-interval", time.Hour*3, "Interval at which the index should be checked for stale series.")
	btIdx.BoolVar(&createCF, "create-cf", true, "enable the creation of the table and column families")

	globalconf.Register("bigtable-idx", btIdx)
	return
}

func ConfigProcess() {
	return
}

func FormatRowKey(id string, partition int32) string {
	return fmt.Sprintf("%d_%s", partition, id)
}
<commit_msg>bigtable_idx.enabled should default to false<commit_after>package bigtable

import (
	"context"
	"flag"
	"fmt"
	"sync"
	"time"

	"cloud.google.com/go/bigtable"
	"github.com/grafana/metrictank/cluster"
	"github.com/grafana/metrictank/idx"
	"github.com/grafana/metrictank/idx/memory"
	"github.com/grafana/metrictank/stats"
	"github.com/raintank/schema"
	"github.com/rakyll/globalconf"
	log "github.com/sirupsen/logrus"
)

const COLUMN_FAMILY = "idx"

var (
	// metric idx.bigtable.query-insert.ok is how many insert queries for a metric completed successfully (triggered by an add or an update)
	statQueryInsertOk = stats.NewCounter32("idx.bigtable.query-insert.ok")
	// metric idx.bigtable.query-insert.fail is how many insert queries for a metric failed (triggered by an add or an update)
	statQueryInsertFail = stats.NewCounter32("idx.bigtable.query-insert.fail")
	// metric idx.bigtable.query-delete.ok is how many delete queries for a metric completed successfully (triggered by an update or a delete)
	statQueryDeleteOk = stats.NewCounter32("idx.bigtable.query-delete.ok")
	// metric idx.bigtable.query-delete.fail is how many delete queries for a metric failed (triggered by an update or a delete)
	statQueryDeleteFail = stats.NewCounter32("idx.bigtable.query-delete.fail")

	// metric idx.bigtable.query-insert.wait is time inserts spent in queue before being executed
	statQueryInsertWaitDuration = stats.NewLatencyHistogram12h32("idx.bigtable.query-insert.wait")
	// metric idx.bigtable.query-insert.exec is time spent executing inserts (possibly repeatedly until success)
	statQueryInsertExecDuration = stats.NewLatencyHistogram15s32("idx.bigtable.query-insert.exec")
	// metric idx.bigtable.query-delete.exec is time spent executing deletes (possibly repeatedly until success)
	statQueryDeleteExecDuration = stats.NewLatencyHistogram15s32("idx.bigtable.query-delete.exec")

	// metric idx.bigtable.add is the duration of an add of one metric to the bigtable idx, including the add to the in-memory index, excluding the insert query
	statAddDuration = stats.NewLatencyHistogram15s32("idx.bigtable.add")
	// metric idx.bigtable.update is the duration of an update of one metric to the bigtable idx, including the update to the in-memory index, excluding any insert/delete queries
	statUpdateDuration = stats.NewLatencyHistogram15s32("idx.bigtable.update")
	// metric idx.bigtable.prune is the duration of a prune of the bigtable idx, including the prune of the in-memory index and all needed delete queries
	statPruneDuration = stats.NewLatencyHistogram15s32("idx.bigtable.prune")
	// metric idx.bigtable.delete is the duration of a delete of one or more metrics from the bigtable idx, including the delete from the in-memory index and the delete query
	statDeleteDuration = stats.NewLatencyHistogram15s32("idx.bigtable.delete")
	// metric idx.bigtable.save.skipped is how many saves have been skipped due to the writeQueue being full
	statSaveSkipped = stats.NewCounter32("idx.bigtable.save.skipped")
	// metric idx.bigtable.save.bytes-per-request is the number of bytes written to bigtable in each request.
	statSaveBytesPerRequest = stats.NewMeter32("idx.bigtable.save.bytes-per-request", true)

	maxStale          time.Duration
	pruneInterval     time.Duration
	updateIdx         bool
	updateInterval    time.Duration
	updateInterval32  uint32
	writeQueueSize    int
	writeConcurrency  int
	writeMaxFlushSize int

	Enabled          bool
	bigtableInstance string
	gcpProject       string
	tableName        string
	createCF         bool
)

func ConfigSetup() {
	btIdx := flag.NewFlagSet("bigtable-idx", flag.ExitOnError)

	btIdx.BoolVar(&Enabled, "enabled", false, "")
	btIdx.StringVar(&gcpProject, "gcp-project", "default", "Name of GCP project the bigtable cluster resides in")
	btIdx.StringVar(&bigtableInstance, "bigtable-instance", "default", "Name of bigtable instance")
	btIdx.StringVar(&tableName, "table-name", "metric_idx", "Name of bigtable table used for metricDefs")
	btIdx.IntVar(&writeQueueSize, "write-queue-size", 100000, "Max number of metricDefs allowed to be unwritten to bigtable. Must be larger then write-max-flush-size")
	btIdx.IntVar(&writeConcurrency, "write-concurrency", 5, "Number of writer threads to use")
	btIdx.IntVar(&writeMaxFlushSize, "write-max-flush-size", 10000, "Max number of metricDefs in each batch write to bigtable")
	btIdx.BoolVar(&updateIdx, "update-bigtable-index", true, "synchronize index changes to bigtable. not all your nodes need to do this.")
	btIdx.DurationVar(&updateInterval, "update-interval", time.Hour*3, "frequency at which we should update the metricDef lastUpdate field, use 0s for instant updates")
	btIdx.DurationVar(&maxStale, "max-stale", 0, "clear series from the index if they have not been seen for this much time.")
	btIdx.DurationVar(&pruneInterval, "prune-interval", time.Hour*3, "Interval at which the index should be checked for stale series.")
	btIdx.BoolVar(&createCF, "create-cf", true, "enable the creation of the table and column families")

	globalconf.Register("bigtable-idx", btIdx)
	return
}

func ConfigProcess() {
	return
}

func FormatRowKey(id string, partition int32) string {
	return fmt.Sprintf("%d_%s", partition, id)
}

{'fraction_non_alphanumeric': 0.09439216439408489, 'fraction_numerical': 0.011907048204340312, 'mean_word_length': 4.944634703196347, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5751867', 'n_tokens_mistral': 3226, 'n_tokens_neox': 3115, 'n_words': 972}