package clusterproxy import ( "encoding/json" "time" "github.com/pingcap/errors" ) const timeFormat = `"2006-01-02 15:04:05.000"` // AdminJobType represents for admin job type, both used in owner and processor type AdminJobType int // FeedState represents for changefeed state type FeedState string type JSONTime time.Time // MarshalJSON used to specify the time format func (t JSONTime) MarshalJSON() ([]byte, error) { stamp := time.Time(t).Format(timeFormat) return []byte(stamp), nil } // UnmarshalJSON is used to parse time.Time from bytes. The builtin json.Unmarshal function cannot unmarshal // a date string formatted as "2006-01-02 15:04:05.000", so we must implement a customized unmarshal function. func (t *JSONTime) UnmarshalJSON(data []byte) error { tm, err := time.Parse(timeFormat, string(data)) if err != nil { return err } *t = JSONTime(tm) return nil } // ListResponse is the response for all List APIs type ListResponse[T any] struct { Total int `json:"total,omitempty"` Items []T `json:"items,omitempty"` } // Tables contains IneligibleTables and EligibleTables type Tables struct { IneligibleTables []TableName `json:"ineligible_tables,omitempty"` EligibleTables []TableName `json:"eligible_tables,omitempty"` } // TableName contains table information type TableName struct { Schema string `json:"database_name,omitempty"` Table string `json:"table_name,omitempty"` TableID int64 `json:"table_id,omitempty"` IsPartition bool `json:"is_partition,omitempty"` } // ResumeChangefeedConfig is used by resume changefeed api type ResumeChangefeedConfig struct { OverwriteCheckpointTs uint64 `json:"overwrite_checkpoint_ts,omitempty"` } // ChangefeedCommonInfo holds some common usage information of a changefeed type ChangefeedCommonInfo struct { ID string `json:"id,omitempty"` FeedState string `json:"state,omitempty"` CheckpointTSO uint64 `json:"checkpoint_tso,omitempty"` RunningError *RunningError `json:"error,omitempty"` } // ChangefeedConfig use by create changefeed api type ChangefeedConfig struct { ID string `json:"changefeed_id,omitempty"` StartTs uint64 `json:"start_ts,omitempty"` TargetTs uint64 `json:"target_ts,omitempty"` SinkURI string `json:"sink_uri,omitempty"` ReplicaConfig *ReplicaConfig `json:"replica_config,omitempty"` } // JSONDuration used to wrap duration into json format type JSONDuration struct { duration time.Duration } // MarshalJSON marshal duration to string func (d JSONDuration) MarshalJSON() ([]byte, error) { return json.Marshal(d.duration.String()) } // UnmarshalJSON unmarshal json value to wrapped duration func (d *JSONDuration) UnmarshalJSON(b []byte) error { var v interface{} if err := json.Unmarshal(b, &v); err != nil { return err } switch value := v.(type) { case float64: d.duration = time.Duration(value) return nil case string: var err error d.duration, err = time.ParseDuration(value) if err != nil { return err } return nil default: return errors.New("invalid duration") } } // ReplicaConfig is a duplicate of config.ReplicaConfig type ReplicaConfig struct { MemoryQuota uint64 `json:"memory_quota,omitempty"` CaseSensitive *bool `json:"case_sensitive,omitempty"` SQLMode string `json:"sql_mode,omitempty"` EnableOldValue *bool `json:"enable_old_value,omitempty"` ForceReplicate *bool `json:"force_replicate,omitempty"` IgnoreIneligibleTable *bool `json:"ignore_ineligible_table,omitempty"` CheckGCSafePoint *bool `json:"check_gc_safe_point,omitempty"` EnableSyncPoint *bool `json:"enable_sync_point,omitempty"` BDRMode *bool `json:"bdr_mode,omitempty"` SyncPointInterval *JSONDuration `json:"sync_point_interval,omitempty"` SyncPointRetention *JSONDuration `json:"sync_point_retention,omitempty"` EnableTableMonitor *bool `json:"enable_table_monitor,omitempty"` ChangefeedErrorStuckDuration uint64 `json:"changefeed_error_stuck_duration,omitempty"` Filter *FilterConfig `json:"filter,omitempty"` Mounter *MounterConfig `json:"mounter,omitempty"` Sink *SinkConfigV2 `json:"sink,omitempty"` Consistent *ConsistentConfig `json:"consistent,omitempty"` Scheduler *ChangefeedSchedulerConfig `json:"scheduler,omitempty"` Integrity *IntegrityConfig `json:"integrity,omitempty"` SyncedStatus *SyncedStatusConfig `json:"synced_status,omitempty"` } // FilterConfig represents filter config for a changefeed // This is a duplicate of config.FilterConfig type FilterConfig struct { *MySQLReplicationRules Rules []string `json:"rules,omitempty"` IgnoreTxnStartTs []uint64 `json:"ignore_txn_start_ts,omitempty"` EventFilters []EventFilterRule `json:"event_filters,omitempty"` } // MounterConfig represents mounter config for a changefeed type MounterConfig struct { WorkerNum int `json:"worker_num,omitempty"` } // EventFilterRule is used by sql event filter and expression filter type EventFilterRule struct { Matcher []string `json:"matcher,omitempty"` IgnoreEvent []string `json:"ignore_event,omitempty"` // regular expression IgnoreSQL []string `json:"ignore_sql,omitempty"` // sql expression IgnoreInsertValueExpr string `json:"ignore_insert_value_expr,omitempty"` IgnoreUpdateNewValueExpr string `json:"ignore_update_new_value_expr,omitempty"` IgnoreUpdateOldValueExpr string `json:"ignore_update_old_value_expr,omitempty"` IgnoreDeleteValueExpr string `json:"ignore_delete_value_expr,omitempty"` } // MySQLReplicationRules is a set of rules based on MySQL's replication tableFilter. type MySQLReplicationRules struct { // DoTables is an allowlist of tables. DoTables []*Table `json:"do_tables,omitempty"` // DoDBs is an allowlist of schemas. DoDBs []string `json:"do_dbs,omitempty"` // IgnoreTables is a blocklist of tables. IgnoreTables []*Table `json:"ignore_tables,omitempty"` // IgnoreDBs is a blocklist of schemas. IgnoreDBs []string `json:"ignore_dbs,omitempty"` } // Table represents a qualified table name. type Table struct { // Schema is the name of the schema (database) containing this table. Schema string `json:"database_name,omitempty"` // Name is the unqualified table name. Name string `json:"table_name,omitempty"` } // DispatchRuleV2 represent the dispatch rules for MQ sink type DispatchRuleV2 struct { Matcher []string `json:"matcher,omitempty"` Topic string `json:"topic,omitempty"` Partition string `json:"partition,omitempty"` IndexName string `json:"index,omitempty"` Columns []string `json:"columns,omitempty"` } // SinkConfigV2 represents sink config for a changefeed // This is a duplicate of config.SinkConfig type SinkConfigV2 struct { Protocol string `json:"protocol,omitempty"` SchemaRegistry string `json:"schema_registry,omitempty"` CSVConfig *CSVConfig `json:"csv,omitempty"` DispatchRules []*DispatchRuleV2 `json:"dispatchers,omitempty"` ColumnSelectors []*ColumnSelector `json:"column_selectors,omitempty"` TxnAtomicity string `json:"transaction_atomicity,omitempty"` EncoderConcurrency int `json:"encoder_concurrency,omitempty"` Terminator string `json:"terminator,omitempty"` DateSeparator string `json:"date_separator,omitempty"` OnlyOutputUpdatedColumns *bool `json:"only_output_updated_columns,omitempty"` CloudStorageConfig *CloudStorageConfig `json:"cloud_storage_config,omitempty"` Open *OpenConfig `json:"open,omitempty"` DebeziumDisableSchema *bool `json:"debezium_disable_schema,omitempty"` DebeziumConfig *DebeziumConfig `json:"debezium,omitempty"` EnablePartitionSeparator *bool `json:"enable_partition_separator,omitempty"` FileIndexWidth int32 `json:"file_index_width,omitempty"` EnableKafkaSinkV2 *bool `json:"enable_kafka_sink_v2,omitempty"` DeleteOnlyOutputHandleKeyColumns *bool `json:"delete_only_output_handle_key_columns,omitempty"` ContentCompatible *bool `json:"content_compatible,omitempty"` SafeMode *bool `json:"safe_mode,omitempty"` AdvanceTimeout int32 `json:"advance_timeout,omitempty"` SendBootstrapIntervalInSec int64 `json:"send_bootstrap_interval_in_sec,omitempty"` SendBootstrapInMsgCount int32 `json:"send_bootstrap_in_msg_count,omitempty"` SendBootstrapToAllPartition *bool `json:"send_bootstrap_to_all_partition,omitempty"` SendAllBootstrapAtStart *bool `json:"send_all_bootstrap_at_start,omitempty"` KafkaSink *KafkaConfig `json:"kafka_config,omitempty"` PulsarConfig *PulsarConfig `json:"pulsar_config,omitempty"` MysqlConfig *MySQLConfig `json:"mysql_config,omitempty"` } // CSVConfig denotes the csv config // This is the same as config.CSVConfig type CSVConfig struct { Delimiter string `json:"delimiter,omitempty"` Quote string `json:"quote,omitempty"` NullString string `json:"null,omitempty"` IncludeCommitTs *bool `json:"include_commit_ts,omitempty"` BinaryEncodingMethod string `json:"binary_encoding_method,omitempty"` OutputOldValue *bool `json:"output_old_value,omitempty"` OutputHandleKey *bool `json:"output_handle_key,omitempty"` OutputFieldHeader *bool `json:"output_field_header,omitempty"` } // ColumnSelector represents a column selector for a table. // This is a duplicate of config.ColumnSelector type ColumnSelector struct { Matcher []string `json:"matcher,omitempty"` Columns []string `json:"columns,omitempty"` } // ConsistentConfig represents replication consistency config for a changefeed // This is a duplicate of config.ConsistentConfig type ConsistentConfig struct { Level string `json:"level,omitempty"` MaxLogSize int64 `json:"max_log_size,omitempty"` FlushIntervalInMs int64 `json:"flush_interval,omitempty"` Storage string `json:"storage,omitempty"` UseFileBackend *bool `json:"use_file_backend,omitempty"` EncodingWorkerNum int32 `json:"encoding_worker_num,omitempty"` FlushWorkerNum int32 `json:"flush_worker_num,omitempty"` Compression string `json:"compression,omitempty"` FlushConcurrency int32 `json:"flush_concurrency,omitempty"` MetaFlushInterval int64 `json:"meta_flush_interval,omitempty"` MemoryUsage *ConsistentMemoryUsage `json:"memory_usage,omitempty"` } type ConsistentMemoryUsage struct { MemoryQuotaPercentage uint64 `json:"memory_quota_percentage,omitempty"` } type ChangefeedSchedulerConfig struct { EnableTableAcrossNodes *bool `json:"enable_table_across_nodes,omitempty"` RegionThreshold int32 `json:"region_threshold,omitempty"` WriteKeyThreshold int32 `json:"write_key_threshold,omitempty"` } type IntegrityConfig struct { IntegrityCheckLevel string `json:"integrity_check_level,omitempty"` CorruptionHandleLevel string `json:"corruption_handle_level,omitempty"` } type SyncedStatusConfig struct { SyncedCheckInterval int64 `json:"synced_check_interval,omitempty"` CheckpointInterval int64 `json:"checkpoint_interval,omitempty"` } type CloudStorageConfig struct { WorkerCount int32 `json:"worker_count,omitempty"` FlushInterval string `json:"flush_interval,omitempty"` FileSize int32 `json:"file_size,omitempty"` FileExpirationDays int32 `json:"file_expiration_days,omitempty"` FileCleanupCronSpec string `json:"file_cleanup_cron_spec,omitempty"` FlushConcurrency int32 `json:"flush_concurrency,omitempty"` OutputRawChangeEvent *bool `json:"output_raw_change_event,omitempty"` OutputColumnId *bool `json:"output_column_id,omitempty"` } type OpenConfig struct { OutputOldValue *bool `json:"output_old_value,omitempty"` } type KafkaConfig struct { PartitionNum int32 `json:"partition_num,omitempty"` ReplicationFactor int32 `json:"replication_factor,omitempty"` KafkaVersion string `json:"kafka_version,omitempty"` MaxMessageBytes int32 `json:"max_message_bytes,omitempty"` Compression string `json:"compression,omitempty"` KafkaClientId string `json:"kafka_client_id,omitempty"` AutoCreateTopic *bool `json:"auto_create_topic,omitempty"` DialTimeout string `json:"dial_timeout,omitempty"` WriteTimeout string `json:"write_timeout,omitempty"` ReadTimeout string `json:"read_timeout,omitempty"` RequiredAcks int32 `json:"required_acks,omitempty"` SaslUser string `json:"sasl_user,omitempty"` SaslPassword string `json:"sasl_password,omitempty"` SaslMechanism string `json:"sasl_mechanism,omitempty"` SaslGssapiAuthType string `json:"sasl_gssapi_auth_type,omitempty"` SaslGssapiKeytabPath string `json:"sasl_gssapi_keytab_path,omitempty"` SaslGssapiKerberosConfigPath string `json:"sasl_gssapi_kerberos_config_path,omitempty"` SaslGssapiServiceName string `json:"sasl_gssapi_service_name,omitempty"` SaslGssapiUser string `json:"sasl_gssapi_user,omitempty"` SaslGssapiPassword string `json:"sasl_gssapi_password,omitempty"` SaslGssapiRealm string `json:"sasl_gssapi_realm,omitempty"` SaslGssapiDisablePafxfast *bool `json:"sasl_gssapi_disable_pafxfast,omitempty"` SaslOauthClientId string `json:"sasl_oauth_client_id,omitempty"` SaslOauthClientSecret string `json:"sasl_oauth_client_secret,omitempty"` SaslOauthTokenUrl string `json:"sasl_oauth_token_url,omitempty"` SaslOauthScopes []string `json:"sasl_oauth_scopes,omitempty"` SaslOauthGrantType string `json:"sasl_oauth_grant_type,omitempty"` SaslOauthAudience string `json:"sasl_oauth_audience,omitempty"` EnableTls *bool `json:"enable_tls,omitempty"` Ca string `json:"ca,omitempty"` Cert string `json:"cert,omitempty"` Key string `json:"key,omitempty"` InsecureSkipVerify *bool `json:"insecure_skip_verify,omitempty"` CodecConfig *CodecConfig `json:"codec_config,omitempty"` LargeMessageHandle *LargeMessageHandleConfig `json:"large_message_handle,omitempty"` GlueSchemaRegistryConfig *GlueSchemaRegistryConfig `json:"glue_schema_registry_config,omitempty"` OutputRawChangeEvent *bool `json:"output_raw_change_event,omitempty"` } type CodecConfig struct { EnableTidbExtension *bool `json:"enable_tidb_extension,omitempty"` MaxBatchSize int32 `json:"max_batch_size,omitempty"` AvroEnableWatermark *bool `json:"avro_enable_watermark,omitempty"` AvroDecimalHandlingMode string `json:"avro_decimal_handling_mode,omitempty"` AvroBigintUnsignedHandlingMode string `json:"avro_bigint_unsigned_handling_mode,omitempty"` EncodingFormat string `json:"encoding_format,omitempty"` } type LargeMessageHandleConfig struct { LargeMessageHandleOption string `json:"large_message_handle_option,omitempty"` LargeMessageHandleCompression string `json:"large_message_handle_compression,omitempty"` ClaimCheckStorageUri string `json:"claim_check_storage_uri,omitempty"` ClaimCheckRawValue *bool `json:"claim_check_raw_value,omitempty"` } type GlueSchemaRegistryConfig struct { RegistryName string `json:"registry_name,omitempty"` Region string `json:"region,omitempty"` AccessKey string `json:"access_key,omitempty"` SecretAccessKey string `json:"secret_access_key,omitempty"` Token string `json:"token,omitempty"` } type PulsarConfig struct { TlsCertificatePath string `json:"tls-certificate-path,omitempty"` TlsPrivateKeyPath string `json:"tls-private-key-path,omitempty"` TlsTrustCertsFilePath string `json:"tls-trust-certs-file-path,omitempty"` PulsarProducerCacheSize int32 `json:"pulsar-producer-cache-size,omitempty"` PulsarVersion string `json:"pulsar-version,omitempty"` CompressionType string `json:"compression-type,omitempty"` AuthenticationToken string `json:"authentication-token,omitempty"` ConnectionTimeout int32 `json:"connection-timeout,omitempty"` OperationTimeout int32 `json:"operation-timeout,omitempty"` BatchingMaxMessages uint32 `json:"batching-max-messages,omitempty"` BatchingMaxPublishDelay int32 `json:"batching-max-publish-delay,omitempty"` SendTimeout int32 `json:"send-timeout,omitempty"` TokenFromFile string `json:"token-from-file,omitempty"` BasicUserName string `json:"basic-user-name,omitempty"` BasicPassword string `json:"basic-password,omitempty"` AuthTlsCertificatePath string `json:"auth-tls-certificate-path,omitempty"` AuthTlsPrivateKeyPath string `json:"auth-tls-private-key-path,omitempty"` Oauth2 *PulsarOAuth2 `json:"oauth2,omitempty"` OutputRawChangeEvent *bool `json:"output-raw-change-event,omitempty"` } type PulsarOAuth2 struct { Oauth2IssuerUrl string `json:"oauth2-issuer-url,omitempty"` Oauth2Audience string `json:"oauth2-audience,omitempty"` Oauth2PrivateKey string `json:"oauth2-private-key,omitempty"` Oauth2ClientId string `json:"oauth2-client-id,omitempty"` Oauth2Scope string `json:"oauth2-scope,omitempty"` } type MySQLConfig struct { WorkerCount int32 `json:"worker_count,omitempty"` MaxTxnRow int32 `json:"max_txn_row,omitempty"` MaxMultiUpdateRowSize int32 `json:"max_multi_update_row_size,omitempty"` MaxMultiUpdateRowCount int32 `json:"max_multi_update_row_count,omitempty"` TidbTxnMode string `json:"tidb_txn_mode,omitempty"` SslCa string `json:"ssl_ca,omitempty"` SslCert string `json:"ssl_cert,omitempty"` SslKey string `json:"ssl_key,omitempty"` TimeZone string `json:"time_zone,omitempty"` WriteTimeout string `json:"write_timeout,omitempty"` ReadTimeout string `json:"read_timeout,omitempty"` Timeout string `json:"timeout,omitempty"` EnableBatchDml *bool `json:"enable_batch_dml,omitempty"` EnableMultiStatement *bool `json:"enable_multi_statement,omitempty"` EnableCachePreparedStatement *bool `json:"enable_cache_prepared_statement,omitempty"` } // RunningError represents some running error from cdc components, // such as processor. type RunningError struct { Addr string `json:"addr,omitempty"` Code string `json:"code,omitempty"` Message string `json:"message,omitempty"` } // ChangeFeedInfo describes the detail of a ChangeFeed type ChangeFeedInfo struct { UpstreamID uint64 `json:"upstream_id,omitempty"` Namespace string `json:"namespace,omitempty"` ID string `json:"id,omitempty"` SinkURI string `json:"sink_uri,omitempty"` CreateTime string `json:"create_time,omitempty"` // Start sync at this commit ts if `StartTs` is specified or using the CreateTime of changefeed. StartTs uint64 `json:"start_ts,omitempty"` // The ChangeFeed will exits until sync to timestamp TargetTs TargetTs uint64 `json:"target_ts,omitempty"` // used for admin job notification, trigger watch event in capture AdminJobType AdminJobType `json:"admin_job_type,omitempty"` Config *ReplicaConfig `json:"config,omitempty"` State FeedState `json:"state,omitempty"` Error *RunningError `json:"error,omitempty"` CreatorVersion string `json:"creator_version,omitempty"` ResolvedTs uint64 `json:"resolved_ts,omitempty"` CheckpointTs uint64 `json:"checkpoint_ts,omitempty"` CheckpointTime string `json:"checkpoint_time,omitempty"` TaskStatus []CaptureTaskStatus `json:"task_status,omitempty"` } // CaptureTaskStatus holds TaskStatus of a capture type CaptureTaskStatus struct { CaptureID string `json:"capture_id,omitempty"` // Table list, containing tables that processor should process Tables []int64 `json:"table_ids,omitempty"` } // DebeziumConfig represents the configurations for debezium protocol encoding type DebeziumConfig struct { OutputOldValue *bool `json:"output_old_value,omitempty"` }