config

Documentation

Index

• Constants
• func AddClickHouseEndpointsOnChange(onChange EndpointsOnChange)
• func CheckCluster(conns common.DBs, clusterName string) error
• func CheckStoragePolicy(conns common.DBs, storagePolicy string) error
• type Auth
• type BaseConfig
• type CKDB
• type CKDBColdStorage
• type CKDiskMonitor
  • func (m *CKDiskMonitor) Validate()
• type CKWriterConfig
• type Config
  • func Load(path string) *Config
  • func (c *Config) GetCKDBColdStorages() map[string]*ckdb.ColdStorage
  • func (c *Config) Validate() error
  • func (c *Config) ValidateAndSetckdbColdStorages() error
• type DatabaseTable
• type Disk
• type DiskCleanup
• type Endpoint
  • func (e Endpoint) String() string
• type EndpointsOnChange
• type HostPort
• type InstanceInfo
• type Location
• type ServerInstanceInfo
  • func NewServerInstranceInfo(ips []net.IP, port, rpcMaxMsgSize int) *ServerInstanceInfo
  • func (p *ServerInstanceInfo) Close()
  • func (p *ServerInstanceInfo) GetNodePodNames() map[string][]string
  • func (p *ServerInstanceInfo) GetServerPodNames() []string
  • func (p *ServerInstanceInfo) Reload() error
  • func (p *ServerInstanceInfo) Start()
• type StorageSetting
• type TraceIdWithIndex
• type Watcher
  • func NewWatcher(cfg *Config, myNodeName, myPodName, myPodNamespace string) (*Watcher, error)
  • func (w *Watcher) GetClickhouseEndpointsWithoutMyself() ([]Endpoint, error)
  • func (w *Watcher) GetMyClickhouseEndpoints() ([]Endpoint, error)
  • func (w *Watcher) Run()

Constants

const (
	DefaultLocalIP                  = "127.0.0.1"
	DefaultControllerPort           = 20035
	DefaultCheckInterval            = 180 // clickhouse是异步删除
	DefaultDiskUsedPercent          = 80
	DefaultDiskFreeSpace            = 300
	DefaultDFDiskPrefix             = "path_"           // In the config.xml of ClickHouse, the disk name of the storage policy 'df_storage' written by deepflow-server starts with 'path_'
	DefaultSystemDiskPrefix         = "default"         // In the config.xml of ClickHouse, the disk name of default storage policy 'default'
	DefaultByconityLocalDiskPrefix  = "server_local_"   // In the config.xml of ByConity, the disk name of the storage policy 'default'
	DefaultBycontiyS3DiskPrefix     = "server_s3_disk_" // In the config.xml of ByConity, the disk name of the storage policy 'cnch_default_s3'
	EnvK8sNodeIP                    = "K8S_NODE_IP_FOR_DEEPFLOW"
	EnvK8sPodName                   = "K8S_POD_NAME_FOR_DEEPFLOW"
	EnvK8sNodeName                  = "K8S_NODE_NAME_FOR_DEEPFLOW"
	EnvK8sNamespace                 = "K8S_NAMESPACE_FOR_DEEPFLOW"
	DefaultCKDBService              = "deepflow-clickhouse"
	DefaultByconityService          = "deepflow-byconity-server"
	DefaultCKDBServicePort          = 9000
	DefaultListenPort               = 20033
	DefaultGrpcBufferSize           = 41943040
	DefaultServiceLabelerLruCap     = 1 << 22
	DefaultCKDBEndpointTCPPortName  = "tcp-port"
	DefaultStatsInterval            = 10      // s
	DefaultFlowTagCacheFlushTimeout = 1800    // s
	DefaultFlowTagCacheMaxSize      = 1 << 18 // 256k
	IndexTypeHash                   = "hash"
	IndexTypeIncremetalIdLocation   = "incremental-id"
	FormatHex                       = "hex"
	FormatDecimal                   = "decimal"
	EnvRunningMode                  = "DEEPFLOW_SERVER_RUNNING_MODE"
	RunningModeStandalone           = "STANDALONE"
	DefaultByconityStoragePolicy    = "cnch_default_s3"
	// the maximum number of endpoints for a server corresponding to ClickHouse;
	//   any endpoints beyond this limit will be ignored
	MaxClickHouseEndpointsPerServer = 128
)

const SYNC_INTERVAL_FAST = 5 * time.Second // When the system first starts, increase the request frequency and obtain Node/Pod information as soon as possible.

const (
	TIMEOUT = 60
)

Functions

func AddClickHouseEndpointsOnChange

func AddClickHouseEndpointsOnChange(onChange EndpointsOnChange)

func CheckCluster

func CheckCluster(conns common.DBs, clusterName string) error

func CheckStoragePolicy

func CheckStoragePolicy(conns common.DBs, storagePolicy string) error

Types

type Auth

type Auth struct {
	Username string `yaml:"username"`
	Password string `yaml:"password"`
}

type BaseConfig

type BaseConfig struct {
	LogFile          string           `yaml:"log-file"`
	LogLevel         string           `yaml:"log-level"`
	TraceIdWithIndex TraceIdWithIndex `yaml:"trace-id-with-index"`
	Base             Config           `yaml:"ingester"`
}

type CKDB

type CKDB struct {
	External bool   `yaml:"external"`
	Type     string `yaml:"type"`
	Host     string `yaml:"host"`

	ActualAddrs         *[]string
	Watcher             *Watcher
	Port                int    `yaml:"port"`
	EndpointTCPPortName string `yaml:"endpoint-tcp-port-name"`
	ClusterName         string `yaml:"cluster-name"`
	StoragePolicy       string `yaml:"storage-policy"`
	TimeZone            string `yaml:"time-zone"`
	// contains filtered or unexported fields
}

type CKDBColdStorage

type CKDBColdStorage struct {
	Enabled  bool             `yaml:"enabled"`
	ColdDisk Disk             `yaml:"cold-disk"`
	Settings []StorageSetting `yaml:"settings,flow"`
}

type CKDiskMonitor

type CKDiskMonitor struct {
	CheckInterval    int             `yaml:"check-interval"` // s
	TTLCheckDisabled bool            `yaml:"ttl-check-disabled"`
	DiskCleanups     []DiskCleanup   `yaml:"disk-cleanups"`
	PriorityDrops    []DatabaseTable `yaml:"priority-drops"`
}

func (*CKDiskMonitor) Validate

func (m *CKDiskMonitor) Validate()

type CKWriterConfig

type CKWriterConfig struct {
	QueueCount   int `yaml:"queue-count"`
	QueueSize    int `yaml:"queue-size"`
	BatchSize    int `yaml:"batch-size"`
	FlushTimeout int `yaml:"flush-timeout"`
}

type Config

type Config struct {
	IsRunningModeStandalone bool
	StorageDisabled         bool            `yaml:"storage-disabled"`
	ListenPort              uint16          `yaml:"listen-port"`
	CKDB                    CKDB            `yaml:"ckdb"`
	ControllerIPs           []string        `yaml:"controller-ips,flow"`
	ControllerPort          uint16          `yaml:"controller-port"`
	CKDBAuth                Auth            `yaml:"ckdb-auth"`
	IngesterEnabled         bool            `yaml:"ingester-enabled"`
	UDPReadBuffer           int             `yaml:"udp-read-buffer"`
	TCPReadBuffer           int             `yaml:"tcp-read-buffer"`
	TCPReaderBuffer         int             `yaml:"tcp-reader-buffer"`
	CKDiskMonitor           CKDiskMonitor   `yaml:"ck-disk-monitor"`
	ColdStorage             CKDBColdStorage `yaml:"ckdb-cold-storage"`

	NodeIP                   string `yaml:"node-ip"`
	GrpcBufferSize           int    `yaml:"grpc-buffer-size"`
	ServiceLabelerLruCap     int    `yaml:"service-labeler-lru-cap"`
	StatsInterval            int    `yaml:"stats-interval"`
	FlowTagCacheFlushTimeout uint32 `yaml:"flow-tag-cache-flush-timeout"`
	FlowTagCacheMaxSize      uint32 `yaml:"flow-tag-cache-max-size"`
	LogFile                  string
	LogLevel                 string
	MyNodeName               string
	TraceIdWithIndex         TraceIdWithIndex
	// contains filtered or unexported fields
}

func Load

func Load(path string) *Config

func (*Config) GetCKDBColdStorages

func (c *Config) GetCKDBColdStorages() map[string]*ckdb.ColdStorage

func (*Config) Validate

func (c *Config) Validate() error

func (*Config) ValidateAndSetckdbColdStorages

func (c *Config) ValidateAndSetckdbColdStorages() error

type DatabaseTable

type DatabaseTable struct {
	Database      string `yaml:"database"`
	TablesContain string `yaml:"tables-contain"`
}

type Disk

type Disk struct {
	Type string `yaml:"type"`
	Name string `yaml:"name"`
}

type DiskCleanup

type DiskCleanup struct {
	DiskNamePrefix string `yaml:"disk-name-prefix"`
	UsedPercent    int    `yaml:"used-percent"` // 0-100
	FreeSpace      int    `yaml:"free-space"`   // Gb
	UsedSpace      int    `yaml:"used-space"`   // Gb
}

type Endpoint

type Endpoint struct {
	Host string
	Port uint16
}

func (Endpoint) String

func (e Endpoint) String() string

type EndpointsOnChange

type EndpointsOnChange interface {
	EndpointsChange([]string)
}

type HostPort

type HostPort struct {
	Host string `yaml:"host"`
	Port string `yaml:"port"`
}

type InstanceInfo

type InstanceInfo struct {
	PodName  string
	NodeName string
}

type Location

type Location struct {
	Start  int    `yaml:"start"`
	Length int    `yaml:"length"`
	Format string `yaml:"format"`
}

type ServerInstanceInfo

type ServerInstanceInfo struct {
	GrpcSession *grpc.GrpcSession
	// contains filtered or unexported fields
}

func NewServerInstranceInfo

func NewServerInstranceInfo(ips []net.IP, port, rpcMaxMsgSize int) *ServerInstanceInfo

func (*ServerInstanceInfo) Close

func (p *ServerInstanceInfo) Close()

func (*ServerInstanceInfo) GetNodePodNames

func (p *ServerInstanceInfo) GetNodePodNames() map[string][]string

func (*ServerInstanceInfo) GetServerPodNames

func (p *ServerInstanceInfo) GetServerPodNames() []string

func (*ServerInstanceInfo) Reload

func (p *ServerInstanceInfo) Reload() error

func (*ServerInstanceInfo) Start

func (p *ServerInstanceInfo) Start()

type StorageSetting

type StorageSetting struct {
	Db        string   `yaml:"db"`
	Tables    []string `yaml:"tables,flow"`
	TTLToMove int      `yaml:"ttl-hour-to-move"`
}

type TraceIdWithIndex

type TraceIdWithIndex struct {
	Disabled              bool     `yaml:"disabled"`
	Type                  string   `yaml:"type"`
	IncrementalIdLocation Location `yaml:"incremental-id-location"`
	FormatIsHex           bool
	TypeIsIncrementalId   bool
}

type Watcher

type Watcher struct {
	NodePodNamesWatch *ServerInstanceInfo
	EndpointWatch     libs.Watcher
	// contains filtered or unexported fields
}

func NewWatcher

func NewWatcher(cfg *Config, myNodeName, myPodName, myPodNamespace string) (*Watcher, error)

func (*Watcher) GetClickhouseEndpointsWithoutMyself

func (w *Watcher) GetClickhouseEndpointsWithoutMyself() ([]Endpoint, error)

func (*Watcher) GetMyClickhouseEndpoints

func (w *Watcher) GetMyClickhouseEndpoints() ([]Endpoint, error)

func (*Watcher) Run

func (w *Watcher) Run()