client.go

// Copyright 2021 TiKV Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

// NOTE: The code in this file is based on code from the
// TiDB project, licensed under the Apache License v 2.0
//
// https://github.com/pingcap/tidb/tree/cc5e161ac06827589c4966674597c137cc9e809c/store/tikv/config/client.go
//

// Copyright 2021 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package config

import (
	"fmt"
	"math"
	"time"

	"google.golang.org/grpc/encoding/gzip"
)

const (
	// DefStoreLivenessTimeout is the default value for store liveness timeout.
	DefStoreLivenessTimeout       = "1s"
	DefGrpcInitialWindowSize      = 1 << 27 // 128MiB
	DefGrpcInitialConnWindowSize  = 1 << 27 // 128MiB
	DefMaxConcurrencyRequestLimit = math.MaxInt64
	DefBatchPolicy                = BatchPolicyStandard
)

const (
	// BatchPolicyBasic is the basic batch policy whose behavior is consistent with versions before v8.3.0.
	BatchPolicyBasic = "basic"
	// BatchPolicyStandard dynamically batches requests based the arrival time intervals of recent requests.
	BatchPolicyStandard = "standard"
	// BatchPolicyPositive always performs additional batching.
	BatchPolicyPositive = "positive"
	// BatchPolicyCustom allows users to customize the internal batch options.
	BatchPolicyCustom = "custom"
)

// TiKVClient is the config for tikv client.
type TiKVClient struct {
	// GrpcConnectionCount is the max gRPC connections that will be established
	// with each tikv-server.
	GrpcConnectionCount uint `toml:"grpc-connection-count" json:"grpc-connection-count"`
	// After a duration of this time in seconds if the client doesn't see any activity it pings
	// the server to see if the transport is still alive.
	GrpcKeepAliveTime uint `toml:"grpc-keepalive-time" json:"grpc-keepalive-time"`
	// After having pinged for keepalive check, the client waits for a duration of Timeout in seconds
	// and if no activity is seen even after that the connection is closed.
	GrpcKeepAliveTimeout float64 `toml:"grpc-keepalive-timeout" json:"grpc-keepalive-timeout"`
	// GrpcCompressionType is the compression type for gRPC channel: none or gzip.
	GrpcCompressionType string `toml:"grpc-compression-type" json:"grpc-compression-type"`
	// GrpcSharedBufferPool is the flag to control whether to share the buffer pool in the TiKV gRPC clients.
	GrpcSharedBufferPool bool `toml:"grpc-shared-buffer-pool" json:"grpc-shared-buffer-pool"`
	// GrpcInitialWindowSize is the value for initial window size on a stream.
	GrpcInitialWindowSize int32 `toml:"grpc-initial-window-size" json:"grpc-initial-window-size"`
	// GrpcInitialConnWindowSize is the value for initial window size on a connection.
	GrpcInitialConnWindowSize int32 `toml:"grpc-initial-conn-window-size" json:"grpc-initial-conn-window-size"`
	// CommitTimeout is the max time which command 'commit' will wait.
	CommitTimeout string      `toml:"commit-timeout" json:"commit-timeout"`
	AsyncCommit   AsyncCommit `toml:"async-commit" json:"async-commit"`

	// BatchPolicy is the policy for batching requests.
	BatchPolicy string `toml:"batch-policy" json:"batch-policy"`
	// MaxBatchSize is the max batch size when calling batch commands API.
	MaxBatchSize uint `toml:"max-batch-size" json:"max-batch-size"`
	// If TiKV load is greater than this, TiDB will wait for a while to avoid little batch.
	OverloadThreshold uint `toml:"overload-threshold" json:"overload-threshold"`
	// MaxBatchWaitTime in nanosecond is the max wait time for batch.
	MaxBatchWaitTime time.Duration `toml:"max-batch-wait-time" json:"max-batch-wait-time"`
	// BatchWaitSize is the max wait size for batch.
	BatchWaitSize uint `toml:"batch-wait-size" json:"batch-wait-size"`
	// EnableChunkRPC indicate the data encode in chunk format for coprocessor requests.
	EnableChunkRPC bool `toml:"enable-chunk-rpc" json:"enable-chunk-rpc"`
	// If a Region has not been accessed for more than the given duration (in seconds), it
	// will be reloaded from the PD.
	RegionCacheTTL uint `toml:"region-cache-ttl" json:"region-cache-ttl"`
	// If a store has been up to the limit, it will return error for successive request to
	// prevent the store occupying too much token in dispatching level.
	StoreLimit int64 `toml:"store-limit" json:"store-limit"`
	// StoreLivenessTimeout is the timeout for store liveness check request.
	StoreLivenessTimeout string           `toml:"store-liveness-timeout" json:"store-liveness-timeout"`
	CoprCache            CoprocessorCache `toml:"copr-cache" json:"copr-cache"`
	// CoprReqTimeout is the timeout for a single coprocessor request
	// Note: this is a transitional modification, and it will be removed if it's dynamic configurable version is ready.
	CoprReqTimeout time.Duration `toml:"copr-req-timeout" json:"copr-req-timeout"`
	// TTLRefreshedTxnSize controls whether a transaction should update its TTL or not.
	TTLRefreshedTxnSize      int64  `toml:"ttl-refreshed-txn-size" json:"ttl-refreshed-txn-size"`
	ResolveLockLiteThreshold uint64 `toml:"resolve-lock-lite-threshold" json:"resolve-lock-lite-threshold"`
	// MaxConcurrencyRequestLimit is the max concurrency number of request to be sent the tikv
	// 0 means auto adjust by feedback.
	MaxConcurrencyRequestLimit int64 `toml:"max-concurrency-request-limit" json:"max-concurrency-request-limit"`
	// EnableReplicaSelectorV2 was deprecated.
	// TODO(crazycs520): remove this config in 8.6 LTS version.
	EnableReplicaSelectorV2 bool `toml:"enable-replica-selector-v2" json:"enable-replica-selector-v2"`

	// RUV2 is the RU v2 TiKV-side weights used to calculate TiKV RU values from ExecDetailsV2.RuV2.
	RUV2 RUV2TiKVConfig `toml:"ru-v2" json:"ru-v2"`

	// TxnChunkWriterAddr is the address of the txn chunk writer for file-based txn.
	TxnChunkWriterAddr string `toml:"txn-chunk-writer-addr" json:"txn-chunk-writer-addr"`
	// TxnChunkWriterConcurrency is the concurrency to request the txn chunk writer for file-based txn.
	TxnChunkWriterConcurrency uint `toml:"txn-chunk-writer-concurrency" json:"txn-chunk-writer-concurrency"`
	// TxnChunkMaxSize is the maximum size of a txn chunk of file-based txn.
	TxnChunkMaxSize uint64 `toml:"txn-chunk-max-size" json:"txn-chunk-max-size"`
	// TxnFileMinMutationSize is the minimum size of mutations to use file-based txn.
	TxnFileMinMutationSize uint64 `toml:"txn-file-min-mutation-size" json:"txn-file-min-mutation-size"`
	// TxnFileRUDiscountRatio is the discount ratio of resource unit for file-based txn.
	// Will be ignored if it's <= 0 or >= 1.
	TxnFileRUDiscountRatio float64 `toml:"txn-file-ru-discount-ratio" json:"txn-file-ru-discount-ratio"`
	// TxnFileRequestSourceWhitelist is the whitelist of request source types (RequestSource.RequestSourceType) that can use file-based txn.
	// For internal requests only. External requests can always use file-based txn.
	TxnFileRequestSourceWhitelist []string `toml:"txn-file-request-source-whitelist" json:"txn-file-request-source-whitelist"`
}

// RUV2TiKVConfig is the configuration for RU v2 TiKV-side weight calculation.
type RUV2TiKVConfig struct {
	// RUScale is the scale factor used to convert RU v2 float values into scaled integer values.
	// It is intentionally chosen to match legacy RU values for compatibility.
	RUScale float64 `toml:"ru-scale" json:"ru-scale"`

	TiKVKVEngineCacheMiss             float64 `toml:"tikv-kv-engine-cache-miss" json:"tikv-kv-engine-cache-miss"`
	ResourceManagerWriteCntTiKV       float64 `toml:"resource-manager-write-cnt-tikv" json:"resource-manager-write-cnt-tikv"`
	ExecutorInputs                    float64 `toml:"executor-inputs" json:"executor-inputs"`
	TiKVCoprocessorExecutorIterations float64 `toml:"tikv-coprocessor-executor-iterations" json:"tikv-coprocessor-executor-iterations"`
	TiKVCoprocessorResponseBytes      float64 `toml:"tikv-coprocessor-response-bytes" json:"tikv-coprocessor-response-bytes"`
	TiKVRaftstoreStoreWriteTriggerWB  float64 `toml:"tikv-raftstore-store-write-trigger-wb-bytes" json:"tikv-raftstore-store-write-trigger-wb-bytes"`
	TiKVStorageProcessedKeysBatchGet  float64 `toml:"tikv-storage-processed-keys-batch-get" json:"tikv-storage-processed-keys-batch-get"`
	TiKVStorageProcessedKeysGet       float64 `toml:"tikv-storage-processed-keys-get" json:"tikv-storage-processed-keys-get"`
}

// DefaultRUV2TiKVConfig returns the default RU v2 TiKV-side configuration.
// These coefficients were fitted from test workloads to map TiKV-side ExecDetailsV2.RuV2 counters
// back to scaled RU values as stably as possible while staying close to the legacy RU v1 accounting.
// Keep RUScale and the per-counter weights in sync with that fitting procedure when updating them.
func DefaultRUV2TiKVConfig() RUV2TiKVConfig {
	return RUV2TiKVConfig{
		RUScale: 2.10,

		TiKVKVEngineCacheMiss:             0.45975389,
		ResourceManagerWriteCntTiKV:       0.09642181,
		ExecutorInputs:                    0.00003150,
		TiKVCoprocessorExecutorIterations: 0.05775369,
		TiKVCoprocessorResponseBytes:      0.00000087,
		TiKVRaftstoreStoreWriteTriggerWB:  0.00006100,
		TiKVStorageProcessedKeysBatchGet:  0.00266791,
		TiKVStorageProcessedKeysGet:       0.01416829,
	}
}

// AsyncCommit is the config for the async commit feature. The switch to enable it is a system variable.
type AsyncCommit struct {
	// Use async commit only if the number of keys does not exceed KeysLimit.
	KeysLimit uint `toml:"keys-limit" json:"keys-limit"`
	// Use async commit only if the total size of keys does not exceed TotalKeySizeLimit.
	TotalKeySizeLimit uint64 `toml:"total-key-size-limit" json:"total-key-size-limit"`
	// The duration within which is safe for async commit or 1PC to commit with an old schema.
	// The following two fields should NOT be modified in most cases. If both async commit
	// and 1PC are disabled in the whole cluster, they can be set to zero to avoid waiting in DDLs.
	SafeWindow time.Duration `toml:"safe-window" json:"safe-window"`
	// The duration in addition to SafeWindow to make DDL safe.
	AllowedClockDrift time.Duration `toml:"allowed-clock-drift" json:"allowed-clock-drift"`
}

// CoprocessorCache is the config for coprocessor cache.
type CoprocessorCache struct {
	// The capacity in MB of the cache. Zero means disable coprocessor cache.
	CapacityMB float64 `toml:"capacity-mb" json:"capacity-mb"`

	// No json fields for below config. Intend to hide them.

	// Only cache requests that containing small number of ranges. May to be changed in future.
	AdmissionMaxRanges uint64 `toml:"admission-max-ranges" json:"-"`
	// Only cache requests whose result set is small.
	AdmissionMaxResultMB float64 `toml:"admission-max-result-mb" json:"-"`
	// Only cache requests takes notable time to process.
	AdmissionMinProcessMs uint64 `toml:"admission-min-process-ms" json:"-"`
}

// DefaultTiKVClient returns default config for TiKVClient.
func DefaultTiKVClient() TiKVClient {
	return TiKVClient{
		GrpcConnectionCount:       4,
		GrpcKeepAliveTime:         10,
		GrpcKeepAliveTimeout:      3,
		GrpcCompressionType:       "none",
		GrpcSharedBufferPool:      false,
		GrpcInitialWindowSize:     DefGrpcInitialWindowSize,
		GrpcInitialConnWindowSize: DefGrpcInitialConnWindowSize,
		CommitTimeout:             "41s",
		AsyncCommit: AsyncCommit{
			// FIXME: Find an appropriate default limit.
			KeysLimit:         256,
			TotalKeySizeLimit: 4 * 1024, // 4 KiB
			SafeWindow:        2 * time.Second,
			AllowedClockDrift: 500 * time.Millisecond,
		},

		BatchPolicy:       DefBatchPolicy,
		MaxBatchSize:      128,
		OverloadThreshold: 200,
		MaxBatchWaitTime:  0,
		BatchWaitSize:     8,

		EnableChunkRPC: true,

		RegionCacheTTL:       600,
		StoreLimit:           0,
		StoreLivenessTimeout: DefStoreLivenessTimeout,

		TTLRefreshedTxnSize: 32 * 1024 * 1024,

		CoprCache: CoprocessorCache{
			CapacityMB:            1000,
			AdmissionMaxRanges:    500,
			AdmissionMaxResultMB:  10,
			AdmissionMinProcessMs: 5,
		},
		CoprReqTimeout: 60 * time.Second,

		ResolveLockLiteThreshold:   512,
		MaxConcurrencyRequestLimit: DefMaxConcurrencyRequestLimit,
		EnableReplicaSelectorV2:    true,
		RUV2:                       DefaultRUV2TiKVConfig(),

		TxnChunkWriterConcurrency:     4,
		TxnChunkMaxSize:               128 * 1024 * 1024,
		TxnFileMinMutationSize:        16 * 1024 * 1024,
		TxnFileRUDiscountRatio:        0.125, // filed-based txn costs 1/8 RU of normal txn.
		TxnFileRequestSourceWhitelist: []string{},
	}
}

// Valid checks if this config is valid.
func (config *TiKVClient) Valid() error {
	if config.GrpcConnectionCount == 0 {
		return fmt.Errorf("grpc-connection-count should be greater than 0")
	}
	if config.GrpcCompressionType != "none" && config.GrpcCompressionType != gzip.Name {
		return fmt.Errorf("grpc-compression-type should be none or %s, but got %s", gzip.Name, config.GrpcCompressionType)
	}
	if config.GetGrpcKeepAliveTimeout() < time.Millisecond*50 {
		return fmt.Errorf("grpc-keepalive-timeout should be at least 0.05, but got %f", config.GrpcKeepAliveTimeout)
	}
	return validateTxnFileConfig(config)
}

func validateTxnFileConfig(config *TiKVClient) error {
	if config.TxnChunkMaxSize == 0 {
		return fmt.Errorf("txn-chunk-max-size should be greater than 0")
	}
	if config.TxnChunkMaxSize > math.MaxInt {
		return fmt.Errorf("txn-chunk-max-size should not exceed %d, but got %d", math.MaxInt, config.TxnChunkMaxSize)
	}
	if config.TxnChunkWriterConcurrency == 0 {
		return fmt.Errorf("txn-chunk-writer-concurrency should be greater than 0")
	}
	if config.TxnChunkWriterConcurrency > math.MaxInt {
		return fmt.Errorf("txn-chunk-writer-concurrency should not exceed %d, but got %d", math.MaxInt, config.TxnChunkWriterConcurrency)
	}
	return nil
}

func (config *TiKVClient) GetGrpcKeepAliveTimeout() time.Duration {
	return time.Duration(config.GrpcKeepAliveTimeout * float64(time.Second))
}