Add configurable traffic cloning

config/README.md

@@ -41,6 +41,9 @@ connection_pool:
   max_idle_conns: 100
   max_idle_conns_per_host: 2
 
+# Optional traffic cloning configuration
+traffic_cloning: <traffic_cloning_config> [optional]
+
 server:
   <server_config> [optional]
 
@@ -157,6 +160,53 @@ params:
     value: <string>
 ```
 
+### <traffic_cloning_config>
+```yml
+# Traffic cloning mirrors accepted requests to one or more additional ClickHouse clusters.
+# If this section is omitted, traffic cloning is disabled and chproxy behaves as before.
+#
+# Setting enable=true without targets is invalid.
+enable: <bool> | default = false [optional]
+
+# Directory for temporary request body captures.
+# If omitted, the operating system default temporary directory is used.
+# The directory is created automatically and must be writable by the chproxy process.
+temp_dir: <string> | optional
+
+# Maximum number of request bodies captured and replayed at the same time.
+# If omitted or zero, there is no limit. When the limit is reached, clone is skipped.
+max_concurrent_captures: <int> | default = 0 [optional]
+
+# Maximum single captured request body size.
+# If omitted or zero, there is no limit. Larger requests are not cloned.
+max_capture_size: <byte_size> | optional
+
+targets:
+  - name: <string>
+
+    # Optional source filters. If omitted, requests from all sources are eligible.
+    # All configured source filters must match for the request to be cloned.
+    from_cluster: <string> | optional
+    from_user: <string> | optional
+    from_cluster_user: <string> | optional
+
+    # Clone destination. The cluster must exist in `clusters`.
+    # The user must exist in the destination cluster users list.
+    to_cluster: <string>
+    to_user: <string>
+
+    # Percentage of matching requests to clone.
+    # Valid range: 0..100. Default: 100.
+    percent: <float> | default = 100 [optional]
+
+    # Optional timeout for the cloned request.
+    timeout: <duration> | optional
+
+    # Deprecated: accepted for backwards compatibility.
+    # Request bodies are captured to temporary files instead.
+    max_queued_chunks: <int> | default = 16 [optional]
+```
+
 ### <server_config>
 ```yml
 # HTTP server configuration

config/config.go

@@ -80,6 +80,8 @@ type Config struct {
 
 	ConnectionPool ConnectionPool `yaml:"connection_pool,omitempty"`
 
+	TrafficCloning TrafficCloning `yaml:"traffic_cloning,omitempty"`
+
 	// Allow to proxy ping requests
 	AllowPing bool `yaml:"allow_ping,omitempty"`
 
@@ -535,6 +537,128 @@ func (c *Proxy) UnmarshalYAML(unmarshal func(interface{}) error) error {
 	return checkOverflow(c.XXX, "proxy")
 }
 
+// TrafficCloning describes optional background cloning of accepted requests
+// to one or more ClickHouse clusters.
+type TrafficCloning struct {
+	// Enable allows enabling cloning without relying on targets being present.
+	Enable bool `yaml:"enable,omitempty"`
+
+	// TempDir is a directory for temporary request body captures.
+	// If omitted, the operating system default temporary directory is used.
+	TempDir string `yaml:"temp_dir,omitempty"`
+
+	// MaxConcurrentCaptures limits the number of request bodies being captured
+	// and replayed to clone targets at the same time.
+	// If omitted or zero, there is no limit.
+	MaxConcurrentCaptures int `yaml:"max_concurrent_captures,omitempty"`
+
+	// MaxConcurrentClones limits the total number of concurrent clone send
+	// goroutines across all targets. If omitted or zero, there is no limit.
+	MaxConcurrentClones int `yaml:"max_concurrent_clones,omitempty"`
+
+	// MaxCaptureSize limits a single captured request body size.
+	// If omitted or zero, there is no size limit.
+	MaxCaptureSize ByteSize `yaml:"max_capture_size,omitempty"`
+
+	// Targets contains clone destinations.
+	Targets []TrafficCloneTarget `yaml:"targets,omitempty"`
+
+	// Catches all undefined fields and must be empty after parsing.
+	XXX map[string]interface{} `yaml:",inline"`
+}
+
+// UnmarshalYAML implements the yaml.Unmarshaler interface.
+func (tc *TrafficCloning) UnmarshalYAML(unmarshal func(interface{}) error) error {
+	type plain TrafficCloning
+	if err := unmarshal((*plain)(tc)); err != nil {
+		return err
+	}
+
+	if tc.Enable && len(tc.Targets) == 0 {
+		return fmt.Errorf("`traffic_cloning.targets` must contain at least one target when traffic cloning is enabled")
+	}
+	if tc.MaxConcurrentCaptures < 0 {
+		return fmt.Errorf("`traffic_cloning.max_concurrent_captures` cannot be negative")
+	}
+	if tc.MaxConcurrentClones < 0 {
+		return fmt.Errorf("`traffic_cloning.max_concurrent_clones` cannot be negative")
+	}
+
+	return checkOverflow(tc.XXX, "traffic_cloning")
+}
+
+func (tc TrafficCloning) IsEnabled() bool {
+	return tc.Enable || len(tc.Targets) > 0
+}
+
+// TrafficCloneTarget describes a single clone destination.
+type TrafficCloneTarget struct {
+	// Name is used in logs and metric labels.
+	Name string `yaml:"name"`
+
+	// FromCluster limits cloning to requests routed to this source cluster.
+	// If omitted, requests from all clusters are eligible.
+	FromCluster string `yaml:"from_cluster,omitempty"`
+
+	// FromUser limits cloning to requests from this chproxy user.
+	// If omitted, requests from all users are eligible.
+	FromUser string `yaml:"from_user,omitempty"`
+
+	// FromClusterUser limits cloning to requests proxied as this source cluster_user.
+	// If omitted, requests from all source cluster users are eligible.
+	FromClusterUser string `yaml:"from_cluster_user,omitempty"`
+
+	// ToCluster is the cluster where cloned requests are sent.
+	ToCluster string `yaml:"to_cluster"`
+
+	// ToUser is the cluster_user from ToCluster used for cloned requests.
+	ToUser string `yaml:"to_user"`
+
+	// Percent controls the sampling rate for this target.
+	// Defaults to 100.
+	Percent float64 `yaml:"percent,omitempty"`
+
+	// Timeout limits clone request execution. Zero means no extra timeout.
+	Timeout Duration `yaml:"timeout,omitempty"`
+
+	// Deprecated: accepted for backwards compatibility with the streaming
+	// implementation. Request bodies are now captured to temporary files.
+	MaxQueuedChunks int `yaml:"max_queued_chunks,omitempty"`
+
+	// Catches all undefined fields and must be empty after parsing.
+	XXX map[string]interface{} `yaml:",inline"`
+}
+
+// UnmarshalYAML implements the yaml.Unmarshaler interface.
+func (t *TrafficCloneTarget) UnmarshalYAML(unmarshal func(interface{}) error) error {
+	*t = TrafficCloneTarget{
+		Percent:         100,
+		MaxQueuedChunks: 16,
+	}
+	type plain TrafficCloneTarget
+	if err := unmarshal((*plain)(t)); err != nil {
+		return err
+	}
+
+	if len(t.Name) == 0 {
+		return fmt.Errorf("`traffic_cloning.target.name` cannot be empty")
+	}
+	if len(t.ToCluster) == 0 {
+		return fmt.Errorf("`traffic_cloning.target.to_cluster` cannot be empty for %q", t.Name)
+	}
+	if len(t.ToUser) == 0 {
+		return fmt.Errorf("`traffic_cloning.target.to_user` cannot be empty for %q", t.Name)
+	}
+	if t.Percent < 0 || t.Percent > 100 {
+		return fmt.Errorf("`traffic_cloning.target.percent` must be between 0 and 100 for %q", t.Name)
+	}
+	if t.MaxQueuedChunks <= 0 {
+		return fmt.Errorf("`traffic_cloning.target.max_queued_chunks` must be positive for %q", t.Name)
+	}
+
+	return checkOverflow(t.XXX, fmt.Sprintf("traffic_cloning.target %q", t.Name))
+}
+
 // Cluster describes CH cluster configuration
 // The simplest configuration consists of:
 //

config/config_test.go

@@ -331,6 +331,86 @@ func TestLoadConfig(t *testing.T) {
 				MaxErrorReasonSize: ByteSize(1 << 50),
 			},
 		},
+		{
+			"traffic cloning",
+			"testdata/traffic_cloning.yml",
+			Config{
+				Server: Server{
+					HTTP: HTTP{
+						ListenAddr: ":8080",
+						TimeoutCfg: TimeoutCfg{
+							ReadTimeout:  Duration(time.Minute),
+							WriteTimeout: Duration(3 * time.Minute),
+							IdleTimeout:  Duration(10 * time.Minute),
+						},
+					},
+				},
+				HackMePlease: true,
+				Clusters: []Cluster{
+					{
+						Name:   "cluster",
+						Scheme: "http",
+						Nodes:  []string{"127.0.0.1:8123"},
+						ClusterUsers: []ClusterUser{
+							{
+								Name: "default",
+							},
+						},
+						HeartBeat: HeartBeat{
+							Interval: Duration(5 * time.Second),
+							Timeout:  Duration(3 * time.Second),
+							Request:  "/ping",
+							Response: "Ok.\n",
+						},
+						RetryNumber: 0,
+					},
+					{
+						Name:   "preprod",
+						Scheme: "http",
+						Nodes:  []string{"127.0.0.2:8123"},
+						ClusterUsers: []ClusterUser{
+							{
+								Name: "default",
+							},
+						},
+						HeartBeat: HeartBeat{
+							Interval: Duration(5 * time.Second),
+							Timeout:  Duration(3 * time.Second),
+							Request:  "/ping",
+							Response: "Ok.\n",
+						},
+						RetryNumber: 0,
+					},
+				},
+				Users: []User{
+					{
+						Name:             "default",
+						ToCluster:        "cluster",
+						ToUser:           "default",
+						MaxExecutionTime: Duration(120 * time.Second),
+					},
+				},
+				TrafficCloning: TrafficCloning{
+					TempDir:               "/tmp/chproxy-traffic-clone",
+					MaxConcurrentCaptures: 4,
+					MaxCaptureSize:        ByteSize(1 << 30),
+					Targets: []TrafficCloneTarget{
+						{
+							Name:            "preprod",
+							FromCluster:     "cluster",
+							FromUser:        "default",
+							FromClusterUser: "default",
+							ToCluster:       "preprod",
+							ToUser:          "default",
+							Percent:         25,
+							Timeout:         Duration(30 * time.Second),
+							MaxQueuedChunks: 32,
+						},
+					},
+				},
+				MaxErrorReasonSize: ByteSize(1 << 50),
+			},
+		},
 	}
 
 	for _, tc := range testCases {
@@ -520,6 +600,11 @@ func TestBadConfig(t *testing.T) {
 			"testdata/bad.max_error_reason_size.yml",
 			"cannot parse byte size \"-10B\": it must be positive float followed by optional units. For example, 1.5Gb, 3T",
 		},
+		{
+			"traffic cloning percent",
+			"testdata/bad.traffic_cloning_percent.yml",
+			"`traffic_cloning.target.percent` must be between 0 and 100 for \"preprod\"",
+		},
 	}
 
 	for _, tc := range testCases {

config/testdata/bad.traffic_cloning_percent.yml

@@ -0,0 +1,21 @@
+server:
+  http:
+    listen_addr: ":8080"
+
+clusters:
+  - name: "cluster"
+    nodes: ["127.0.0.1:8123"]
+    users:
+      - name: "default"
+
+users:
+  - name: "default"
+    to_cluster: "cluster"
+    to_user: "default"
+
+traffic_cloning:
+  targets:
+    - name: "preprod"
+      to_cluster: "cluster"
+      to_user: "default"
+      percent: 101

config/testdata/traffic_cloning.yml

@@ -0,0 +1,36 @@
+server:
+  http:
+    listen_addr: ":8080"
+
+hack_me_please: true
+
+clusters:
+  - name: "cluster"
+    nodes: ["127.0.0.1:8123"]
+    users:
+      - name: "default"
+
+  - name: "preprod"
+    nodes: ["127.0.0.2:8123"]
+    users:
+      - name: "default"
+
+users:
+  - name: "default"
+    to_cluster: "cluster"
+    to_user: "default"
+
+traffic_cloning:
+  temp_dir: "/tmp/chproxy-traffic-clone"
+  max_concurrent_captures: 4
+  max_capture_size: 1GB
+  targets:
+    - name: "preprod"
+      from_cluster: "cluster"
+      from_user: "default"
+      from_cluster_user: "default"
+      to_cluster: "preprod"
+      to_user: "default"
+      percent: 25
+      timeout: 30s
+      max_queued_chunks: 32

docs/src/content/docs/configuration/metrics.md

@@ -33,9 +33,12 @@ Metrics are exposed in [Prometheus text format](https://prometheus.io/docs/instr
 | response_body_bytes_total | Counter | The amount of bytes written to response bodies | `user`, `cluster`, `cluster_user`, `replica`, `cluster_node` |
 | status_codes_total | Counter | Distribution by response status codes | `user`, `cluster`, `cluster_user`, `replica`, `cluster_node`, `code` |
 | timeout_request_total | Counter | The number of timed out requests | `user`, `cluster`, `cluster_user`, `replica`, `cluster_node` |
+| traffic_clone_duration_seconds | Summary | Traffic clone request duration | `user`, `cluster`, `cluster_user`, `target`, `target_cluster`, `target_cluster_user` |
+| traffic_clone_in_flight | Gauge | The number of traffic clone requests in flight | `user`, `cluster`, `cluster_user`, `target`, `target_cluster`, `target_cluster_user` |
+| traffic_clone_requests_total | Counter | The number of traffic clone requests by result | `user`, `cluster`, `cluster_user`, `target`, `target_cluster`, `target_cluster_user`, `result` |
+| traffic_clone_status_codes_total | Counter | Distribution by status codes returned by traffic clone requests | `user`, `cluster`, `cluster_user`, `target`, `target_cluster`, `target_cluster_user`, `code` |
 | user_queue_overflow_total | Counter | The number of overflows for per-user request queues | `user`, `cluster`, `cluster_user` |
 
 An example of [Grafana's](https://grafana.com) dashboard for `chproxy` metrics is available [here](https://github.com/ContentSquare/chproxy/blob/master/chproxy_overview.json).
 
 ![dashboard example](https://user-images.githubusercontent.com/2902918/31392734-b2fd4a18-ade2-11e7-84a9-4aaaac4c10d7.png)
-