Compare commits

..

1 Commits

Author SHA1 Message Date
jianghang.loujh
1a00be1aad fixed format 2024-09-11 15:38:20 +08:00
94 changed files with 192 additions and 1474 deletions

View File

@@ -3,7 +3,7 @@
<parent> <parent>
<artifactId>canal.admin</artifactId> <artifactId>canal.admin</artifactId>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
</parent> </parent>
<modelVersion>4.0.0</modelVersion> <modelVersion>4.0.0</modelVersion>

View File

@@ -3,7 +3,7 @@
<parent> <parent>
<artifactId>canal.admin</artifactId> <artifactId>canal.admin</artifactId>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
</parent> </parent>
<modelVersion>4.0.0</modelVersion> <modelVersion>4.0.0</modelVersion>
<artifactId>canal.admin-web</artifactId> <artifactId>canal.admin-web</artifactId>

View File

@@ -11,8 +11,8 @@ SET FOREIGN_KEY_CHECKS = 0;
DROP TABLE IF EXISTS `canal_adapter_config`; DROP TABLE IF EXISTS `canal_adapter_config`;
CREATE TABLE `canal_adapter_config` ( CREATE TABLE `canal_adapter_config` (
`id` bigint(20) NOT NULL AUTO_INCREMENT, `id` bigint(20) NOT NULL AUTO_INCREMENT,
`category` varchar(255) NOT NULL, `category` varchar(45) NOT NULL,
`name` varchar(255) NOT NULL, `name` varchar(45) NOT NULL,
`status` varchar(45) DEFAULT NULL, `status` varchar(45) DEFAULT NULL,
`content` text NOT NULL, `content` text NOT NULL,
`modified_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, `modified_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
@@ -25,7 +25,7 @@ CREATE TABLE `canal_adapter_config` (
DROP TABLE IF EXISTS `canal_cluster`; DROP TABLE IF EXISTS `canal_cluster`;
CREATE TABLE `canal_cluster` ( CREATE TABLE `canal_cluster` (
`id` bigint(20) NOT NULL AUTO_INCREMENT, `id` bigint(20) NOT NULL AUTO_INCREMENT,
`name` varchar(255) NOT NULL, `name` varchar(63) NOT NULL,
`zk_hosts` varchar(255) NOT NULL, `zk_hosts` varchar(255) NOT NULL,
`modified_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, `modified_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`id`) PRIMARY KEY (`id`)
@@ -39,7 +39,7 @@ CREATE TABLE `canal_config` (
`id` bigint(20) NOT NULL AUTO_INCREMENT, `id` bigint(20) NOT NULL AUTO_INCREMENT,
`cluster_id` bigint(20) DEFAULT NULL, `cluster_id` bigint(20) DEFAULT NULL,
`server_id` bigint(20) DEFAULT NULL, `server_id` bigint(20) DEFAULT NULL,
`name` varchar(255) NOT NULL, `name` varchar(45) NOT NULL,
`status` varchar(45) DEFAULT NULL, `status` varchar(45) DEFAULT NULL,
`content` text NOT NULL, `content` text NOT NULL,
`content_md5` varchar(128) NOT NULL, `content_md5` varchar(128) NOT NULL,
@@ -56,7 +56,7 @@ CREATE TABLE `canal_instance_config` (
`id` bigint(20) NOT NULL AUTO_INCREMENT, `id` bigint(20) NOT NULL AUTO_INCREMENT,
`cluster_id` bigint(20) DEFAULT NULL, `cluster_id` bigint(20) DEFAULT NULL,
`server_id` bigint(20) DEFAULT NULL, `server_id` bigint(20) DEFAULT NULL,
`name` varchar(255) NOT NULL, `name` varchar(45) NOT NULL,
`status` varchar(45) DEFAULT NULL, `status` varchar(45) DEFAULT NULL,
`content` text NOT NULL, `content` text NOT NULL,
`content_md5` varchar(128) DEFAULT NULL, `content_md5` varchar(128) DEFAULT NULL,
@@ -72,8 +72,8 @@ DROP TABLE IF EXISTS `canal_node_server`;
CREATE TABLE `canal_node_server` ( CREATE TABLE `canal_node_server` (
`id` bigint(20) NOT NULL AUTO_INCREMENT, `id` bigint(20) NOT NULL AUTO_INCREMENT,
`cluster_id` bigint(20) DEFAULT NULL, `cluster_id` bigint(20) DEFAULT NULL,
`name` varchar(255) NOT NULL, `name` varchar(63) NOT NULL,
`ip` varchar(255) NOT NULL, `ip` varchar(63) NOT NULL,
`admin_port` int(11) DEFAULT NULL, `admin_port` int(11) DEFAULT NULL,
`tcp_port` int(11) DEFAULT NULL, `tcp_port` int(11) DEFAULT NULL,
`metric_port` int(11) DEFAULT NULL, `metric_port` int(11) DEFAULT NULL,
@@ -88,10 +88,10 @@ CREATE TABLE `canal_node_server` (
DROP TABLE IF EXISTS `canal_user`; DROP TABLE IF EXISTS `canal_user`;
CREATE TABLE `canal_user` ( CREATE TABLE `canal_user` (
`id` bigint(20) NOT NULL AUTO_INCREMENT, `id` bigint(20) NOT NULL AUTO_INCREMENT,
`username` varchar(255) NOT NULL, `username` varchar(31) NOT NULL,
`password` varchar(255) NOT NULL, `password` varchar(128) NOT NULL,
`name` varchar(255) NOT NULL, `name` varchar(31) NOT NULL,
`roles` varchar(255) NOT NULL, `roles` varchar(31) NOT NULL,
`introduction` varchar(255) DEFAULT NULL, `introduction` varchar(255) DEFAULT NULL,
`avatar` varchar(255) DEFAULT NULL, `avatar` varchar(255) DEFAULT NULL,
`creation_date` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, `creation_date` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,

View File

@@ -3,13 +3,13 @@
<parent> <parent>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId> <artifactId>canal</artifactId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<modelVersion>4.0.0</modelVersion> <modelVersion>4.0.0</modelVersion>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<artifactId>canal.admin</artifactId> <artifactId>canal.admin</artifactId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<packaging>pom</packaging> <packaging>pom</packaging>
<name>canal admin module for otter ${project.version}</name> <name>canal admin module for otter ${project.version}</name>
<modules> <modules>

View File

@@ -1,160 +0,0 @@
# helm charts 使用说明
包含`canal-admin`和`canal-server`两个chart,基本配置保存在k8s的`secret`里,`canal.properties`和`instance.properties`保存在mysql数据库中,使用zookeeper注册中心实现高可用。
- [helm charts 使用说明](#helm-charts-使用说明)
- [创建zk集群](#创建zk集群)
- [初始化数据库](#初始化数据库)
- [部署canal-admin chart](#部署canal-admin-chart)
- [配置values.yaml](#配置valuesyaml)
- [配置集群信息](#配置集群信息)
- [部署canal-server chart](#部署canal-server-chart)
- [配置values.yaml](#配置valuesyaml-1)
- [instance 实例创建](#instance-实例创建)
## 创建zk集群
```sh
helm install canal-zookeeper oci://registry-1.docker.io/bitnamicharts/zookeeper
```
如需挂载磁盘参考 [bitnami/zookeeper](`https://github.com/bitnami/charts/tree/main/bitnami/zookeeper`)
## 初始化数据库
[/admin/admin-web/src/main/resources/canal_manager.sql](/admin/admin-web/src/main/resources/canal_manager.sql)
[/deployer/src/main/resources/spring/tsdb/sql/create_table.sql](/deployer/src/main/resources/spring/tsdb/sql/create_table.sql)
## 部署canal-admin chart
### 配置values.yaml
```yaml
# 主要配置
admin:
config: |
server:
port: 8089
spring:
jackson:
date-format: yyyy-MM-dd HH:mm:ss
time-zone: GMT+8
spring.datasource:
address: your_db_host:3306
database: canal_manager
username: ****
password: ****
driver-class-name: com.mysql.jdbc.Driver
url: jdbc:mysql://${spring.datasource.address}/${spring.datasource.database}?useUnicode=true&characterEncoding=UTF-8&useSSL=false&allowPublicKeyRetrieval=true
hikari:
maximum-pool-size: 30
minimum-idle: 1
canal:
adminUser: admin
adminPasswd: admin
# debug: true
```
全量参数见:[canal-admin/values.yaml](./canal-admin/values.yaml),测试用例见:[tests/admin-values.yaml](./tests/admin-values.yaml)
```sh
helm install canal-admin -f ./admin-values.yaml ./canal-admin
```
### 配置集群信息
![cluster](./images/cluster.png)
等待pod启动后,使用默认账户`admin`密码`123456`登录
创建集群`default`并配置`zk地址`为`canal-zookeeper.default.svc.cluster.local:2181`
配置完成后,点击`配置`-`主配置`,修改以下配置项,不是覆盖整个文件
```yaml
canal.user = admin
canal.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441
canal.zkServers = canal-zookeeper.default.svc.cluster.local:2181
# tcp, kafka, rocketMQ, rabbitMQ, pulsarMQ 看实际需求,此处改为了rabbitMQ
canal.serverMode = rabbitMQ
#canal.instance.tsdb.dir = ${canal.file.data.dir:../conf}/${canal.instance.destination:}
canal.instance.tsdb.url = jdbc:mysql://you_db_host:3306/canal_tsdb?useUnicode=true&characterEncoding=UTF-8&useSSL=false
canal.instance.tsdb.dbUsername = ****
canal.instance.tsdb.dbPassword = ****
#canal.instance.tsdb.spring.xml = classpath:spring/tsdb/h2-tsdb.xml
canal.instance.tsdb.spring.xml = classpath:spring/tsdb/mysql-tsdb.xml
#canal.instance.global.spring.xml = classpath:spring/file-instance.xml
canal.instance.global.spring.xml = classpath:spring/default-instance.xml
# 如果用的是阿里云polardb需要配置
canal.aliyun.accessKey = ****
canal.aliyun.secretKey = ****
canal.aliyun.uid = 主账号id
# 我们是推送到rabbitmq所以需要配置
rabbitmq.host = ****:5672
rabbitmq.virtual.host = ****
rabbitmq.exchange = ****
# canal.aliyun.accessKey
rabbitmq.username = ****
# canal.aliyun.secretKey
rabbitmq.password = ****
rabbitmq.deliveryMode = direct
```
## 部署canal-server chart
### 配置values.yaml
```yaml
# 主要配置
server:
config: |
canal.port = 11111
canal.metrics.pull.port = 11112
# register ip
canal.register.ip =
# canal admin config
canal.admin.manager = canal-admin.default:8089
canal.admin.port = 11110
canal.admin.user = admin
canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441
# admin auto register
canal.admin.register.auto = true
canal.admin.register.cluster = default
```
```sh
helm install canal-server -f ./server-values.yaml ./canal-server
```
全量参数见:[canal-server/values.yaml](./canal-server/values.yaml),测试用例见:[tests/server-values.yaml](./tests/server-values.yaml)
![server](./images/server.png)
至此,canal-server已自动创建,并自动注册。
## instance 实例创建
主要配置如下
```yaml
# 配置你要监听的数据库
canal.instance.master.address=127.0.0.1:3306
canal.instance.dbUsername=****
canal.instance.dbPassword=****
# 过滤监听哪些表
canal.instance.filter.regex=.*\\..*
# 如果是推送到rabbitmq,需要配置 Routing Key
canal.mq.topic=你的Routing Key
```

View File

@@ -1,23 +0,0 @@
# Patterns to ignore when building packages.
# This supports shell glob matching, relative path matching, and
# negation (prefixed with !). Only one pattern per line.
.DS_Store
# Common VCS dirs
.git/
.gitignore
.bzr/
.bzrignore
.hg/
.hgignore
.svn/
# Common backup files
*.swp
*.bak
*.tmp
*.orig
*~
# Various IDEs
.project
.idea/
*.tmproj
.vscode/

View File

@@ -1,24 +0,0 @@
apiVersion: v2
name: canal-admin
description: A Helm chart for Kubernetes
# A chart can be either an 'application' or a 'library' chart.
#
# Application charts are a collection of templates that can be packaged into versioned archives
# to be deployed.
#
# Library charts provide useful utilities or functions for the chart developer. They're included as
# a dependency of application charts to inject those utilities and functions into the rendering
# pipeline. Library charts do not define any templates and therefore cannot be deployed.
type: application
# This is the chart version. This version number should be incremented each time you make changes
# to the chart and its templates, including the app version.
# Versions are expected to follow Semantic Versioning (https://semver.org/)
version: 1.1.8-alpha-3
# This is the version number of the application being deployed. This version number should be
# incremented each time you make changes to the application. Versions are not expected to
# follow Semantic Versioning. They should reflect the version the application is using.
# It is recommended to use it with quotes.
appVersion: "1.16.0"

View File

@@ -1,22 +0,0 @@
1. Get the application URL by running these commands:
{{- if .Values.ingress.enabled }}
{{- range $host := .Values.ingress.hosts }}
{{- range .paths }}
http{{ if $.Values.ingress.tls }}s{{ end }}://{{ $host.host }}{{ .path }}
{{- end }}
{{- end }}
{{- else if contains "NodePort" .Values.service.type }}
export NODE_PORT=$(kubectl get --namespace {{ .Release.Namespace }} -o jsonpath="{.spec.ports[0].nodePort}" services {{ include "canal-admin.fullname" . }})
export NODE_IP=$(kubectl get nodes --namespace {{ .Release.Namespace }} -o jsonpath="{.items[0].status.addresses[0].address}")
echo http://$NODE_IP:$NODE_PORT
{{- else if contains "LoadBalancer" .Values.service.type }}
NOTE: It may take a few minutes for the LoadBalancer IP to be available.
You can watch its status by running 'kubectl get --namespace {{ .Release.Namespace }} svc -w {{ include "canal-admin.fullname" . }}'
export SERVICE_IP=$(kubectl get svc --namespace {{ .Release.Namespace }} {{ include "canal-admin.fullname" . }} --template "{{"{{ range (index .status.loadBalancer.ingress 0) }}{{.}}{{ end }}"}}")
echo http://$SERVICE_IP:{{ .Values.service.port }}
{{- else if contains "ClusterIP" .Values.service.type }}
export POD_NAME=$(kubectl get pods --namespace {{ .Release.Namespace }} -l "app.kubernetes.io/name={{ include "canal-admin.name" . }},app.kubernetes.io/instance={{ .Release.Name }}" -o jsonpath="{.items[0].metadata.name}")
export CONTAINER_PORT=$(kubectl get pod --namespace {{ .Release.Namespace }} $POD_NAME -o jsonpath="{.spec.containers[0].ports[0].containerPort}")
echo "Visit http://127.0.0.1:8080 to use your application"
kubectl --namespace {{ .Release.Namespace }} port-forward $POD_NAME 8080:$CONTAINER_PORT
{{- end }}

View File

@@ -1,62 +0,0 @@
{{/*
Expand the name of the chart.
*/}}
{{- define "canal-admin.name" -}}
{{- default .Chart.Name .Values.nameOverride | trunc 63 | trimSuffix "-" }}
{{- end }}
{{/*
Create a default fully qualified app name.
We truncate at 63 chars because some Kubernetes name fields are limited to this (by the DNS naming spec).
If release name contains chart name it will be used as a full name.
*/}}
{{- define "canal-admin.fullname" -}}
{{- if .Values.fullnameOverride }}
{{- .Values.fullnameOverride | trunc 63 | trimSuffix "-" }}
{{- else }}
{{- $name := default .Chart.Name .Values.nameOverride }}
{{- if contains $name .Release.Name }}
{{- .Release.Name | trunc 63 | trimSuffix "-" }}
{{- else }}
{{- printf "%s-%s" .Release.Name $name | trunc 63 | trimSuffix "-" }}
{{- end }}
{{- end }}
{{- end }}
{{/*
Create chart name and version as used by the chart label.
*/}}
{{- define "canal-admin.chart" -}}
{{- printf "%s-%s" .Chart.Name .Chart.Version | replace "+" "_" | trunc 63 | trimSuffix "-" }}
{{- end }}
{{/*
Common labels
*/}}
{{- define "canal-admin.labels" -}}
helm.sh/chart: {{ include "canal-admin.chart" . }}
{{ include "canal-admin.selectorLabels" . }}
{{- if .Chart.AppVersion }}
app.kubernetes.io/version: {{ .Chart.AppVersion | quote }}
{{- end }}
app.kubernetes.io/managed-by: {{ .Release.Service }}
{{- end }}
{{/*
Selector labels
*/}}
{{- define "canal-admin.selectorLabels" -}}
app.kubernetes.io/name: {{ include "canal-admin.name" . }}
app.kubernetes.io/instance: {{ .Release.Name }}
{{- end }}
{{/*
Create the name of the service account to use
*/}}
{{- define "canal-admin.serviceAccountName" -}}
{{- if .Values.serviceAccount.create }}
{{- default (include "canal-admin.fullname" .) .Values.serviceAccount.name }}
{{- else }}
{{- default "default" .Values.serviceAccount.name }}
{{- end }}
{{- end }}

View File

@@ -1,75 +0,0 @@
apiVersion: apps/v1
kind: Deployment
metadata:
name: {{ include "canal-admin.fullname" . }}
labels:
{{- include "canal-admin.labels" . | nindent 4 }}
spec:
{{- if not .Values.autoscaling.enabled }}
replicas: {{ .Values.replicaCount }}
{{- end }}
selector:
matchLabels:
{{- include "canal-admin.selectorLabels" . | nindent 6 }}
template:
metadata:
{{- with .Values.podAnnotations }}
annotations:
{{- toYaml . | nindent 8 }}
{{- end }}
labels:
{{- include "canal-admin.labels" . | nindent 8 }}
{{- with .Values.podLabels }}
{{- toYaml . | nindent 8 }}
{{- end }}
spec:
{{- with .Values.imagePullSecrets }}
imagePullSecrets:
{{- toYaml . | nindent 8 }}
{{- end }}
serviceAccountName: {{ include "canal-admin.serviceAccountName" . }}
securityContext:
{{- toYaml .Values.podSecurityContext | nindent 8 }}
containers:
- name: {{ .Chart.Name }}
securityContext:
{{- toYaml .Values.securityContext | nindent 12 }}
image: "{{ .Values.image.repository }}:{{ .Values.image.tag | default .Chart.AppVersion }}"
imagePullPolicy: {{ .Values.image.pullPolicy }}
ports:
- name: http
containerPort: {{ .Values.service.port }}
protocol: TCP
livenessProbe:
{{- toYaml .Values.livenessProbe | nindent 12 }}
readinessProbe:
{{- toYaml .Values.readinessProbe | nindent 12 }}
resources:
{{- toYaml .Values.resources | nindent 12 }}
volumeMounts:
- mountPath: /home/admin/canal-admin/conf/application.yml
name: secret
subPath: application.yml
{{- with .Values.volumeMounts }}
{{- toYaml . | nindent 12 }}
{{- end }}
volumes:
- name: secret
secret:
defaultMode: 420
secretName: {{ include "canal-admin.fullname" . }}
{{- with .Values.volumes }}
{{- toYaml . | nindent 8 }}
{{- end }}
{{- with .Values.nodeSelector }}
nodeSelector:
{{- toYaml . | nindent 8 }}
{{- end }}
{{- with .Values.affinity }}
affinity:
{{- toYaml . | nindent 8 }}
{{- end }}
{{- with .Values.tolerations }}
tolerations:
{{- toYaml . | nindent 8 }}
{{- end }}

View File

@@ -1,32 +0,0 @@
{{- if .Values.autoscaling.enabled }}
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
name: {{ include "canal-admin.fullname" . }}
labels:
{{- include "canal-admin.labels" . | nindent 4 }}
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: {{ include "canal-admin.fullname" . }}
minReplicas: {{ .Values.autoscaling.minReplicas }}
maxReplicas: {{ .Values.autoscaling.maxReplicas }}
metrics:
{{- if .Values.autoscaling.targetCPUUtilizationPercentage }}
- type: Resource
resource:
name: cpu
target:
type: Utilization
averageUtilization: {{ .Values.autoscaling.targetCPUUtilizationPercentage }}
{{- end }}
{{- if .Values.autoscaling.targetMemoryUtilizationPercentage }}
- type: Resource
resource:
name: memory
target:
type: Utilization
averageUtilization: {{ .Values.autoscaling.targetMemoryUtilizationPercentage }}
{{- end }}
{{- end }}

View File

@@ -1,43 +0,0 @@
{{- if .Values.ingress.enabled -}}
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
name: {{ include "canal-admin.fullname" . }}
labels:
{{- include "canal-admin.labels" . | nindent 4 }}
{{- with .Values.ingress.annotations }}
annotations:
{{- toYaml . | nindent 4 }}
{{- end }}
spec:
{{- with .Values.ingress.className }}
ingressClassName: {{ . }}
{{- end }}
{{- if .Values.ingress.tls }}
tls:
{{- range .Values.ingress.tls }}
- hosts:
{{- range .hosts }}
- {{ . | quote }}
{{- end }}
secretName: {{ .secretName }}
{{- end }}
{{- end }}
rules:
{{- range .Values.ingress.hosts }}
- host: {{ .host | quote }}
http:
paths:
{{- range .paths }}
- path: {{ .path }}
{{- with .pathType }}
pathType: {{ . }}
{{- end }}
backend:
service:
name: {{ include "canal-admin.fullname" $ }}
port:
number: {{ $.Values.service.port }}
{{- end }}
{{- end }}
{{- end }}

View File

@@ -1,9 +0,0 @@
apiVersion: v1
kind: Secret
metadata:
name: {{ include "canal-admin.fullname" . }}
labels:
{{- include "canal-admin.labels" . | nindent 4 }}
type: Opaque
data:
application.yml: {{tpl .Values.admin.config $ | b64enc | quote }}

View File

@@ -1,15 +0,0 @@
apiVersion: v1
kind: Service
metadata:
name: {{ include "canal-admin.fullname" . }}
labels:
{{- include "canal-admin.labels" . | nindent 4 }}
spec:
type: {{ .Values.service.type }}
ports:
- port: {{ .Values.service.port }}
targetPort: http
protocol: TCP
name: http
selector:
{{- include "canal-admin.selectorLabels" . | nindent 4 }}

View File

@@ -1,13 +0,0 @@
{{- if .Values.serviceAccount.create -}}
apiVersion: v1
kind: ServiceAccount
metadata:
name: {{ include "canal-admin.serviceAccountName" . }}
labels:
{{- include "canal-admin.labels" . | nindent 4 }}
{{- with .Values.serviceAccount.annotations }}
annotations:
{{- toYaml . | nindent 4 }}
{{- end }}
automountServiceAccountToken: {{ .Values.serviceAccount.automount }}
{{- end }}

View File

@@ -1,15 +0,0 @@
apiVersion: v1
kind: Pod
metadata:
name: "{{ include "canal-admin.fullname" . }}-test-connection"
labels:
{{- include "canal-admin.labels" . | nindent 4 }}
annotations:
"helm.sh/hook": test
spec:
containers:
- name: wget
image: busybox
command: ['wget']
args: ['{{ include "canal-admin.fullname" . }}:{{ .Values.service.port }}']
restartPolicy: Never

View File

@@ -1,146 +0,0 @@
# Default values for canal-admin.
# This is a YAML-formatted file.
# Declare variables to be passed into your templates.
# This will set the replicaset count more information can be found here: https://kubernetes.io/docs/concepts/workloads/controllers/replicaset/
replicaCount: 1
# This sets the container image more information can be found here: https://kubernetes.io/docs/concepts/containers/images/
image:
repository: canal/canal-admin
# This sets the pull policy for images.
pullPolicy: IfNotPresent
# Overrides the image tag whose default is the chart appVersion.
tag: "latest"
# This is for the secretes for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/
imagePullSecrets: []
# This is to override the chart name.
nameOverride: ""
fullnameOverride: ""
# This section builds out the service account more information can be found here: https://kubernetes.io/docs/concepts/security/service-accounts/
serviceAccount:
# Specifies whether a service account should be created
create: false
# Automatically mount a ServiceAccount's API credentials?
automount: false
# Annotations to add to the service account
annotations: {}
# The name of the service account to use.
# If not set and create is true, a name is generated using the fullname template
name: ""
# This is for setting Kubernetes Annotations to a Pod.
# For more information checkout: https://kubernetes.io/docs/concepts/overview/working-with-objects/annotations/
podAnnotations: {}
# This is for setting Kubernetes Labels to a Pod.
# For more information checkout: https://kubernetes.io/docs/concepts/overview/working-with-objects/labels/
podLabels: {}
podSecurityContext: {}
# fsGroup: 2000
securityContext: {}
# capabilities:
# drop:
# - ALL
# readOnlyRootFilesystem: true
# runAsNonRoot: true
# runAsUser: 1000
# This is for setting up a service more information can be found here: https://kubernetes.io/docs/concepts/services-networking/service/
service:
# This sets the service type more information can be found here: https://kubernetes.io/docs/concepts/services-networking/service/#publishing-services-service-types
type: ClusterIP
# This sets the ports more information can be found here: https://kubernetes.io/docs/concepts/services-networking/service/#field-spec-ports
port: 8089
# This block is for setting up the ingress for more information can be found here: https://kubernetes.io/docs/concepts/services-networking/ingress/
ingress:
enabled: false
className: ""
annotations: {}
# kubernetes.io/ingress.class: nginx
# kubernetes.io/tls-acme: "true"
hosts:
- host: chart-example.local
paths:
- path: /
pathType: ImplementationSpecific
tls: []
# - secretName: chart-example-tls
# hosts:
# - chart-example.local
resources: {}
# We usually recommend not to specify default resources and to leave this as a conscious
# choice for the user. This also increases chances charts run on environments with little
# resources, such as Minikube. If you do want to specify resources, uncomment the following
# lines, adjust them as necessary, and remove the curly braces after 'resources:'.
# limits:
# cpu: 100m
# memory: 128Mi
# requests:
# cpu: 100m
# memory: 128Mi
# This is to setup the liveness and readiness probes more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/configure-liveness-readiness-startup-probes/
livenessProbe:
httpGet:
path: /
port: http
readinessProbe:
httpGet:
path: /
port: http
# This section is for setting up autoscaling more information can be found here: https://kubernetes.io/docs/concepts/workloads/autoscaling/
autoscaling:
enabled: false
minReplicas: 1
maxReplicas: 100
targetCPUUtilizationPercentage: 80
# targetMemoryUtilizationPercentage: 80
# Additional volumes on the output Deployment definition.
volumes: []
# - name: foo
# secret:
# secretName: mysecret
# optional: false
# Additional volumeMounts on the output Deployment definition.
volumeMounts: []
# - name: foo
# mountPath: "/etc/foo"
# readOnly: true
nodeSelector: {}
tolerations: []
affinity: {}
admin:
config: |
server:
port: 8089
spring:
jackson:
date-format: yyyy-MM-dd HH:mm:ss
time-zone: GMT+8
spring.datasource:
address: you_db_host:3306
database: canal_manager
username: ****
password: ****
driver-class-name: com.mysql.jdbc.Driver
url: jdbc:mysql://${spring.datasource.address}/${spring.datasource.database}?useUnicode=true&characterEncoding=UTF-8&useSSL=false&allowPublicKeyRetrieval=true
hikari:
maximum-pool-size: 30
minimum-idle: 1
canal:
adminUser: admin
adminPasswd: admin
# debug: true

View File

@@ -1,23 +0,0 @@
# Patterns to ignore when building packages.
# This supports shell glob matching, relative path matching, and
# negation (prefixed with !). Only one pattern per line.
.DS_Store
# Common VCS dirs
.git/
.gitignore
.bzr/
.bzrignore
.hg/
.hgignore
.svn/
# Common backup files
*.swp
*.bak
*.tmp
*.orig
*~
# Various IDEs
.project
.idea/
*.tmproj
.vscode/

View File

@@ -1,24 +0,0 @@
apiVersion: v2
name: canal-server
description: A Helm chart for Kubernetes
# A chart can be either an 'application' or a 'library' chart.
#
# Application charts are a collection of templates that can be packaged into versioned archives
# to be deployed.
#
# Library charts provide useful utilities or functions for the chart developer. They're included as
# a dependency of application charts to inject those utilities and functions into the rendering
# pipeline. Library charts do not define any templates and therefore cannot be deployed.
type: application
# This is the chart version. This version number should be incremented each time you make changes
# to the chart and its templates, including the app version.
# Versions are expected to follow Semantic Versioning (https://semver.org/)
version: 1.1.8-alpha-3
# This is the version number of the application being deployed. This version number should be
# incremented each time you make changes to the application. Versions are not expected to
# follow Semantic Versioning. They should reflect the version the application is using.
# It is recommended to use it with quotes.
appVersion: "1.16.0"

View File

@@ -1,22 +0,0 @@
1. Get the application URL by running these commands:
{{- if .Values.ingress.enabled }}
{{- range $host := .Values.ingress.hosts }}
{{- range .paths }}
http{{ if $.Values.ingress.tls }}s{{ end }}://{{ $host.host }}{{ .path }}
{{- end }}
{{- end }}
{{- else if contains "NodePort" .Values.service.type }}
export NODE_PORT=$(kubectl get --namespace {{ .Release.Namespace }} -o jsonpath="{.spec.ports[0].nodePort}" services {{ include "canal-server.fullname" . }})
export NODE_IP=$(kubectl get nodes --namespace {{ .Release.Namespace }} -o jsonpath="{.items[0].status.addresses[0].address}")
echo http://$NODE_IP:$NODE_PORT
{{- else if contains "LoadBalancer" .Values.service.type }}
NOTE: It may take a few minutes for the LoadBalancer IP to be available.
You can watch its status by running 'kubectl get --namespace {{ .Release.Namespace }} svc -w {{ include "canal-server.fullname" . }}'
export SERVICE_IP=$(kubectl get svc --namespace {{ .Release.Namespace }} {{ include "canal-server.fullname" . }} --template "{{"{{ range (index .status.loadBalancer.ingress 0) }}{{.}}{{ end }}"}}")
echo http://$SERVICE_IP:{{ .Values.service.port }}
{{- else if contains "ClusterIP" .Values.service.type }}
export POD_NAME=$(kubectl get pods --namespace {{ .Release.Namespace }} -l "app.kubernetes.io/name={{ include "canal-server.name" . }},app.kubernetes.io/instance={{ .Release.Name }}" -o jsonpath="{.items[0].metadata.name}")
export CONTAINER_PORT=$(kubectl get pod --namespace {{ .Release.Namespace }} $POD_NAME -o jsonpath="{.spec.containers[0].ports[0].containerPort}")
echo "Visit http://127.0.0.1:8080 to use your application"
kubectl --namespace {{ .Release.Namespace }} port-forward $POD_NAME 8080:$CONTAINER_PORT
{{- end }}

View File

@@ -1,62 +0,0 @@
{{/*
Expand the name of the chart.
*/}}
{{- define "canal-server.name" -}}
{{- default .Chart.Name .Values.nameOverride | trunc 63 | trimSuffix "-" }}
{{- end }}
{{/*
Create a default fully qualified app name.
We truncate at 63 chars because some Kubernetes name fields are limited to this (by the DNS naming spec).
If release name contains chart name it will be used as a full name.
*/}}
{{- define "canal-server.fullname" -}}
{{- if .Values.fullnameOverride }}
{{- .Values.fullnameOverride | trunc 63 | trimSuffix "-" }}
{{- else }}
{{- $name := default .Chart.Name .Values.nameOverride }}
{{- if contains $name .Release.Name }}
{{- .Release.Name | trunc 63 | trimSuffix "-" }}
{{- else }}
{{- printf "%s-%s" .Release.Name $name | trunc 63 | trimSuffix "-" }}
{{- end }}
{{- end }}
{{- end }}
{{/*
Create chart name and version as used by the chart label.
*/}}
{{- define "canal-server.chart" -}}
{{- printf "%s-%s" .Chart.Name .Chart.Version | replace "+" "_" | trunc 63 | trimSuffix "-" }}
{{- end }}
{{/*
Common labels
*/}}
{{- define "canal-server.labels" -}}
helm.sh/chart: {{ include "canal-server.chart" . }}
{{ include "canal-server.selectorLabels" . }}
{{- if .Chart.AppVersion }}
app.kubernetes.io/version: {{ .Chart.AppVersion | quote }}
{{- end }}
app.kubernetes.io/managed-by: {{ .Release.Service }}
{{- end }}
{{/*
Selector labels
*/}}
{{- define "canal-server.selectorLabels" -}}
app.kubernetes.io/name: {{ include "canal-server.name" . }}
app.kubernetes.io/instance: {{ .Release.Name }}
{{- end }}
{{/*
Create the name of the service account to use
*/}}
{{- define "canal-server.serviceAccountName" -}}
{{- if .Values.serviceAccount.create }}
{{- default (include "canal-server.fullname" .) .Values.serviceAccount.name }}
{{- else }}
{{- default "default" .Values.serviceAccount.name }}
{{- end }}
{{- end }}

View File

@@ -1,32 +0,0 @@
{{- if .Values.autoscaling.enabled }}
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
name: {{ include "canal-server.fullname" . }}
labels:
{{- include "canal-server.labels" . | nindent 4 }}
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: {{ include "canal-server.fullname" . }}
minReplicas: {{ .Values.autoscaling.minReplicas }}
maxReplicas: {{ .Values.autoscaling.maxReplicas }}
metrics:
{{- if .Values.autoscaling.targetCPUUtilizationPercentage }}
- type: Resource
resource:
name: cpu
target:
type: Utilization
averageUtilization: {{ .Values.autoscaling.targetCPUUtilizationPercentage }}
{{- end }}
{{- if .Values.autoscaling.targetMemoryUtilizationPercentage }}
- type: Resource
resource:
name: memory
target:
type: Utilization
averageUtilization: {{ .Values.autoscaling.targetMemoryUtilizationPercentage }}
{{- end }}
{{- end }}

View File

@@ -1,9 +0,0 @@
apiVersion: v1
kind: Secret
metadata:
name: {{ include "canal-server.fullname" . }}
labels:
{{- include "canal-server.labels" . | nindent 4 }}
type: Opaque
data:
canal.properties: {{tpl .Values.server.config $ | b64enc | quote }}

View File

@@ -1,23 +0,0 @@
apiVersion: v1
kind: Service
metadata:
name: {{ printf "%s-headless" (include "canal-server.fullname" .) | trunc 63 | trimSuffix "-" }}
labels:
{{- include "canal-server.labels" . | nindent 4 }}
spec:
type: {{ .Values.service.type }}
ports:
- port: {{ .Values.service.ports.admin }}
targetPort: admin
protocol: TCP
name: tcp-admin
- port: {{ .Values.service.ports.tcp }}
targetPort: tcp
protocol: TCP
name: tcp-tcp
- port: {{ .Values.service.ports.metric }}
targetPort: metric
protocol: TCP
name: tcp-metric
selector:
{{- include "canal-server.selectorLabels" . | nindent 4 }}

View File

@@ -1,13 +0,0 @@
{{- if .Values.serviceAccount.create -}}
apiVersion: v1
kind: ServiceAccount
metadata:
name: {{ include "canal-server.serviceAccountName" . }}
labels:
{{- include "canal-server.labels" . | nindent 4 }}
{{- with .Values.serviceAccount.annotations }}
annotations:
{{- toYaml . | nindent 4 }}
{{- end }}
automountServiceAccountToken: {{ .Values.serviceAccount.automount }}
{{- end }}

View File

@@ -1,80 +0,0 @@
apiVersion: apps/v1
kind: StatefulSet
metadata:
name: {{ include "canal-server.fullname" . }}
labels:
{{- include "canal-server.labels" . | nindent 4 }}
spec:
serviceName: {{ printf "%s-headless" (include "canal-server.fullname" .) | trunc 63 | trimSuffix "-" }}
replicas: {{ .Values.replicaCount }}
selector:
matchLabels:
{{- include "canal-server.selectorLabels" . | nindent 6 }}
template:
metadata:
{{- with .Values.podAnnotations }}
annotations:
{{- toYaml . | nindent 8 }}
{{- end }}
labels:
{{- include "canal-server.labels" . | nindent 8 }}
{{- with .Values.podLabels }}
{{- toYaml . | nindent 8 }}
{{- end }}
spec:
{{- with .Values.imagePullSecrets }}
imagePullSecrets:
{{- toYaml . | nindent 8 }}
{{- end }}
serviceAccountName: {{ include "canal-server.serviceAccountName" . }}
securityContext:
{{- toYaml .Values.podSecurityContext | nindent 8 }}
containers:
- name: {{ .Chart.Name }}
securityContext:
{{- toYaml .Values.securityContext | nindent 12 }}
image: "{{ .Values.image.repository }}:{{ .Values.image.tag | default .Chart.AppVersion }}"
imagePullPolicy: {{ .Values.image.pullPolicy }}
ports:
- name: admin
containerPort: {{ .Values.service.ports.admin }}
protocol: TCP
- name: tcp
containerPort: {{ .Values.service.ports.tcp }}
protocol: TCP
- name: metric
containerPort: {{ .Values.service.ports.metric }}
protocol: TCP
livenessProbe:
{{- toYaml .Values.livenessProbe | nindent 12 }}
readinessProbe:
{{- toYaml .Values.readinessProbe | nindent 12 }}
resources:
{{- toYaml .Values.resources | nindent 12 }}
volumeMounts:
- mountPath: /home/admin/canal-server/conf/canal.properties
name: secret
subPath: canal.properties
{{- with .Values.volumeMounts }}
{{- toYaml . | nindent 12 }}
{{- end }}
volumes:
- name: secret
secret:
defaultMode: 420
secretName: {{ include "canal-server.fullname" . }}
{{- with .Values.volumes }}
{{- toYaml . | nindent 8 }}
{{- end }}
{{- with .Values.nodeSelector }}
nodeSelector:
{{- toYaml . | nindent 8 }}
{{- end }}
{{- with .Values.affinity }}
affinity:
{{- toYaml . | nindent 8 }}
{{- end }}
{{- with .Values.tolerations }}
tolerations:
{{- toYaml . | nindent 8 }}
{{- end }}

View File

@@ -1,15 +0,0 @@
apiVersion: v1
kind: Pod
metadata:
name: "{{ include "canal-server.fullname" . }}-test-connection"
labels:
{{- include "canal-server.labels" . | nindent 4 }}
annotations:
"helm.sh/hook": test
spec:
containers:
- name: wget
image: busybox
command: ['wget']
args: ['{{ include "canal-server.fullname" . }}:{{ .Values.service.port }}']
restartPolicy: Never

View File

@@ -1,144 +0,0 @@
# Default values for canal-server.
# This is a YAML-formatted file.
# Declare variables to be passed into your templates.
# This will set the replicaset count more information can be found here: https://kubernetes.io/docs/concepts/workloads/controllers/replicaset/
replicaCount: 1
# This sets the container image more information can be found here: https://kubernetes.io/docs/concepts/containers/images/
image:
repository: canal/canal-server
# This sets the pull policy for images.
pullPolicy: IfNotPresent
# Overrides the image tag whose default is the chart appVersion.
tag: "1.1.8-alpha-3"
# This is for the secretes for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/
imagePullSecrets: []
# This is to override the chart name.
nameOverride: ""
fullnameOverride: ""
# This section builds out the service account more information can be found here: https://kubernetes.io/docs/concepts/security/service-accounts/
serviceAccount:
# Specifies whether a service account should be created
create: false
# Automatically mount a ServiceAccount's API credentials?
automount: false
# Annotations to add to the service account
annotations: {}
# The name of the service account to use.
# If not set and create is true, a name is generated using the fullname template
name: ""
# This is for setting Kubernetes Annotations to a Pod.
# For more information checkout: https://kubernetes.io/docs/concepts/overview/working-with-objects/annotations/
podAnnotations: {}
# This is for setting Kubernetes Labels to a Pod.
# For more information checkout: https://kubernetes.io/docs/concepts/overview/working-with-objects/labels/
podLabels: {}
podSecurityContext: {}
# fsGroup: 2000
securityContext: {}
# capabilities:
# drop:
# - ALL
# readOnlyRootFilesystem: true
# runAsNonRoot: true
# runAsUser: 1000
# This is for setting up a service more information can be found here: https://kubernetes.io/docs/concepts/services-networking/service/
service:
# This sets the service type more information can be found here: https://kubernetes.io/docs/concepts/services-networking/service/#publishing-services-service-types
type: ClusterIP
# This sets the ports more information can be found here: https://kubernetes.io/docs/concepts/services-networking/service/#field-spec-ports
# port: 80
ports:
admin: 11110
tcp: 11111
metric: 11112
# This block is for setting up the ingress for more information can be found here: https://kubernetes.io/docs/concepts/services-networking/ingress/
ingress:
enabled: false
className: ""
annotations: {}
# kubernetes.io/ingress.class: nginx
# kubernetes.io/tls-acme: "true"
hosts:
- host: chart-example.local
paths:
- path: /
pathType: ImplementationSpecific
tls: []
# - secretName: chart-example-tls
# hosts:
# - chart-example.local
resources: {}
# We usually recommend not to specify default resources and to leave this as a conscious
# choice for the user. This also increases chances charts run on environments with little
# resources, such as Minikube. If you do want to specify resources, uncomment the following
# lines, adjust them as necessary, and remove the curly braces after 'resources:'.
# limits:
# cpu: 100m
# memory: 128Mi
# requests:
# cpu: 100m
# memory: 128Mi
# This is to setup the liveness and readiness probes more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/configure-liveness-readiness-startup-probes/
livenessProbe:
httpGet:
path: /
port: metric
readinessProbe:
httpGet:
path: /
port: metric
# This section is for setting up autoscaling more information can be found here: https://kubernetes.io/docs/concepts/workloads/autoscaling/
autoscaling:
enabled: false
minReplicas: 1
maxReplicas: 100
targetCPUUtilizationPercentage: 80
# targetMemoryUtilizationPercentage: 80
# Additional volumes on the output Deployment definition.
volumes: []
# - name: foo
# secret:
# secretName: mysecret
# optional: false
# Additional volumeMounts on the output Deployment definition.
volumeMounts: []
# - name: foo
# mountPath: "/etc/foo"
# readOnly: true
nodeSelector: {}
tolerations: []
affinity: {}
server:
config: |
canal.port = 11111
canal.metrics.pull.port = 11112
# register ip
canal.register.ip =
# canal admin config
canal.admin.manager = canal-admin.default:8089
canal.admin.port = 11110
canal.admin.user = admin
canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441
# admin auto register
canal.admin.register.auto = true
canal.admin.register.cluster = default

Binary file not shown.

Before

Width:  |  Height:  |  Size: 100 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 157 KiB

View File

@@ -1,34 +0,0 @@
ingress:
enabled: true
hosts:
- host: admin.canal.com
paths:
- path: /
pathType: ImplementationSpecific
admin:
config: |
server:
port: 8089
spring:
jackson:
date-format: yyyy-MM-dd HH:mm:ss
time-zone: GMT+8
spring.datasource:
address: you_db_host:3306
database: canal_manager
username: ****
password: ****
driver-class-name: com.mysql.jdbc.Driver
url: jdbc:mysql://${spring.datasource.address}/${spring.datasource.database}?useUnicode=true&characterEncoding=UTF-8&useSSL=false&allowPublicKeyRetrieval=true
hikari:
maximum-pool-size: 30
minimum-idle: 1
canal:
adminUser: admin
adminPasswd: admin
debug: true
image:
repository: registry.us-east-1.aliyuncs.com/sll0/canal-admin
tag: "latest"

View File

@@ -1,20 +0,0 @@
server:
config: |
canal.port = 11111
canal.metrics.pull.port = 11112
# register ip
canal.register.ip =
# canal admin config
canal.admin.manager = canal-admin.default:8089
canal.admin.port = 11110
canal.admin.user = admin
canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441
# admin auto register
canal.admin.register.auto = true
canal.admin.register.cluster = default
image:
repository: registry.us-east-1.aliyuncs.com/sll0/canal-server
tag: "v1.1.8-alpha-3"

View File

@@ -1,5 +0,0 @@
#! /bin/bash
helm install canal-admin -f ./admin-values.yaml ../canal-admin --dry-run
helm install canal-server -f ./server-values.yaml ../canal-server --dry-run

View File

@@ -1,9 +1,11 @@
<?xml version="1.0" encoding="UTF-8"?> <?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent> <parent>
<artifactId>canal.client-adapter</artifactId> <artifactId>canal.client-adapter</artifactId>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<modelVersion>4.0.0</modelVersion> <modelVersion>4.0.0</modelVersion>

View File

@@ -40,7 +40,7 @@ public class ClickHouseEtlService extends AbstractEtlService {
public EtlResult importData(List<String> params) { public EtlResult importData(List<String> params) {
DbMapping dbMapping = config.getDbMapping(); DbMapping dbMapping = config.getDbMapping();
DruidDataSource dataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey()); DruidDataSource dataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
String sql = "SELECT * FROM " + SyncUtil.getSourceDbTableName(dbMapping, dataSource.getDbType()); String sql = "SELECT * FROM " + SyncUtil.getDbTableName(dbMapping, dataSource.getDbType());
return importData(sql, params); return importData(sql, params);
} }
@@ -126,13 +126,13 @@ public class ClickHouseEtlService extends AbstractEtlService {
int i = 1; int i = 1;
for (Map.Entry<String, String> entry : columnsMap.entrySet()) { for (Map.Entry<String, String> entry : columnsMap.entrySet()) {
String targetColumnName = entry.getKey(); String targetClolumnName = entry.getKey();
String srcColumnName = entry.getValue(); String srcColumnName = entry.getValue();
if (srcColumnName == null) { if (srcColumnName == null) {
srcColumnName = targetColumnName; srcColumnName = targetClolumnName;
} }
Integer type = columnType.get(targetColumnName.toLowerCase()); Integer type = columnType.get(targetClolumnName.toLowerCase());
Object value = rs.getObject(srcColumnName); Object value = rs.getObject(srcColumnName);
if (value != null) { if (value != null) {
SyncUtil.setPStmt(type, pstmt, value, i); SyncUtil.setPStmt(type, pstmt, value, i);

View File

@@ -268,17 +268,6 @@ public class SyncUtil {
return result; return result;
} }
public static String getSourceDbTableName(MappingConfig.DbMapping dbMapping, String dbType) {
String result = "";
String backtick = getBacktickByDbType(dbType);
if (StringUtils.isNotEmpty(dbMapping.getDatabase())) {
result += (backtick + dbMapping.getDatabase() + backtick + ".");
}
result += (backtick + dbMapping.getTable() + backtick);
return result;
}
/** /**
* 根据DbType返回反引号或空字符串 * 根据DbType返回反引号或空字符串
* *

View File

@@ -3,7 +3,7 @@
<parent> <parent>
<artifactId>canal.client-adapter</artifactId> <artifactId>canal.client-adapter</artifactId>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<modelVersion>4.0.0</modelVersion> <modelVersion>4.0.0</modelVersion>

View File

@@ -3,7 +3,7 @@
<parent> <parent>
<artifactId>canal.client-adapter</artifactId> <artifactId>canal.client-adapter</artifactId>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<modelVersion>4.0.0</modelVersion> <modelVersion>4.0.0</modelVersion>

View File

@@ -3,7 +3,7 @@
<parent> <parent>
<artifactId>canal.client-adapter</artifactId> <artifactId>canal.client-adapter</artifactId>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<modelVersion>4.0.0</modelVersion> <modelVersion>4.0.0</modelVersion>

View File

@@ -3,7 +3,7 @@
<parent> <parent>
<artifactId>canal.client-adapter</artifactId> <artifactId>canal.client-adapter</artifactId>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<modelVersion>4.0.0</modelVersion> <modelVersion>4.0.0</modelVersion>

View File

@@ -3,7 +3,7 @@
<parent> <parent>
<artifactId>canal.client-adapter</artifactId> <artifactId>canal.client-adapter</artifactId>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<modelVersion>4.0.0</modelVersion> <modelVersion>4.0.0</modelVersion>

View File

@@ -3,7 +3,7 @@
<parent> <parent>
<artifactId>canal.client-adapter</artifactId> <artifactId>canal.client-adapter</artifactId>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<modelVersion>4.0.0</modelVersion> <modelVersion>4.0.0</modelVersion>

View File

@@ -3,7 +3,7 @@
<parent> <parent>
<artifactId>canal.client-adapter</artifactId> <artifactId>canal.client-adapter</artifactId>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<modelVersion>4.0.0</modelVersion> <modelVersion>4.0.0</modelVersion>

View File

@@ -3,7 +3,7 @@
<parent> <parent>
<artifactId>canal.client-adapter</artifactId> <artifactId>canal.client-adapter</artifactId>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<modelVersion>4.0.0</modelVersion> <modelVersion>4.0.0</modelVersion>

View File

@@ -217,7 +217,7 @@ public class CommonRest {
resStatus = "off"; resStatus = "off";
} }
Map<String, String> res = new LinkedHashMap<>(); Map<String, String> res = new LinkedHashMap<>();
res.put("status", resStatus); res.put("stauts", resStatus);
return res; return res;
} }
} }

View File

@@ -3,7 +3,7 @@
<parent> <parent>
<artifactId>canal.client-adapter</artifactId> <artifactId>canal.client-adapter</artifactId>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<modelVersion>4.0.0</modelVersion> <modelVersion>4.0.0</modelVersion>

View File

@@ -3,7 +3,7 @@
<parent> <parent>
<artifactId>canal.client-adapter</artifactId> <artifactId>canal.client-adapter</artifactId>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<modelVersion>4.0.0</modelVersion> <modelVersion>4.0.0</modelVersion>

View File

@@ -352,13 +352,13 @@ public class PhoenixEtlService {
Map<String, Object> insertValues = new HashMap<>(); Map<String, Object> insertValues = new HashMap<>();
int i = 1; int i = 1;
for (Map.Entry<String, String> entry : columnsMap.entrySet()) { for (Map.Entry<String, String> entry : columnsMap.entrySet()) {
String targetColumnName = entry.getKey(); String targetClolumnName = entry.getKey();
String srcColumnName = entry.getValue(); String srcColumnName = entry.getValue();
if (srcColumnName == null) { if (srcColumnName == null) {
srcColumnName = targetColumnName; srcColumnName = targetClolumnName;
} }
Integer type = columnType.get(targetColumnName.toLowerCase()); Integer type = columnType.get(targetClolumnName.toLowerCase());
try { try {
Object value = rs.getObject(srcColumnName); Object value = rs.getObject(srcColumnName);

View File

@@ -4,12 +4,12 @@
<parent> <parent>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId> <artifactId>canal</artifactId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<artifactId>canal.client-adapter</artifactId> <artifactId>canal.client-adapter</artifactId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<packaging>pom</packaging> <packaging>pom</packaging>
<name>canal client adapter module for otter ${project.version}</name> <name>canal client adapter module for otter ${project.version}</name>
@@ -136,7 +136,7 @@
<artifactId>kafka-clients</artifactId> <artifactId>kafka-clients</artifactId>
<version>${kafka_version}</version> <version>${kafka_version}</version>
</dependency> </dependency>
<!-- rabbitmq --> <!-- rabbotmq -->
<dependency> <dependency>
<groupId>com.rabbitmq</groupId> <groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId> <artifactId>amqp-client</artifactId>

View File

@@ -3,7 +3,7 @@
<parent> <parent>
<artifactId>canal.client-adapter</artifactId> <artifactId>canal.client-adapter</artifactId>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<modelVersion>4.0.0</modelVersion> <modelVersion>4.0.0</modelVersion>

View File

@@ -39,7 +39,7 @@ public class RdbEtlService extends AbstractEtlService {
DbMapping dbMapping = config.getDbMapping(); DbMapping dbMapping = config.getDbMapping();
// 获取源数据源,根据数据库类型拼装数据库名和表名 // 获取源数据源,根据数据库类型拼装数据库名和表名
DruidDataSource dataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey()); DruidDataSource dataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey());
String sql = "SELECT * FROM " + SyncUtil.getSourceDbTableName(dbMapping, dataSource.getDbType()); String sql = "SELECT * FROM " + SyncUtil.getDbTableName(dbMapping, dataSource.getDbType());
return importData(sql, params); return importData(sql, params);
} }
@@ -122,13 +122,13 @@ public class RdbEtlService extends AbstractEtlService {
int i = 1; int i = 1;
for (Map.Entry<String, String> entry : columnsMap.entrySet()) { for (Map.Entry<String, String> entry : columnsMap.entrySet()) {
String targetColumnName = entry.getKey(); String targetClolumnName = entry.getKey();
String srcColumnName = entry.getValue(); String srcColumnName = entry.getValue();
if (srcColumnName == null) { if (srcColumnName == null) {
srcColumnName = targetColumnName; srcColumnName = targetClolumnName;
} }
Integer type = columnType.get(targetColumnName.toLowerCase()); Integer type = columnType.get(targetClolumnName.toLowerCase());
Object value = rs.getObject(srcColumnName); Object value = rs.getObject(srcColumnName);
if (value != null) { if (value != null) {
SyncUtil.setPStmt(type, pstmt, value, i); SyncUtil.setPStmt(type, pstmt, value, i);

View File

@@ -284,17 +284,6 @@ public class SyncUtil {
return result; return result;
} }
public static String getSourceDbTableName(MappingConfig.DbMapping dbMapping, String dbType) {
String result = "";
String backtick = getBacktickByDbType(dbType);
if (StringUtils.isNotEmpty(dbMapping.getDatabase())) {
result += (backtick + dbMapping.getDatabase() + backtick + ".");
}
result += (backtick + dbMapping.getTable() + backtick);
return result;
}
/** /**
* 根据DbType返回反引号或空字符串 * 根据DbType返回反引号或空字符串
* *

View File

@@ -3,7 +3,7 @@
<parent> <parent>
<artifactId>canal.client-adapter</artifactId> <artifactId>canal.client-adapter</artifactId>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<modelVersion>4.0.0</modelVersion> <modelVersion>4.0.0</modelVersion>

View File

@@ -3,7 +3,7 @@
<parent> <parent>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId> <artifactId>canal</artifactId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>

View File

@@ -1,14 +1,15 @@
package com.alibaba.otter.canal.client.impl; package com.alibaba.otter.canal.client.impl;
import java.net.SocketAddress;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.client.CanalConnector; import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalNodeAccessStrategy; import com.alibaba.otter.canal.client.CanalNodeAccessStrategy;
import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.exception.CanalClientException; import com.alibaba.otter.canal.protocol.exception.CanalClientException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.net.SocketAddress;
import java.util.concurrent.TimeUnit;
/** /**
* 集群版本connector实现,自带了failover功能<br/> * 集群版本connector实现,自带了failover功能<br/>
@@ -64,7 +65,7 @@ public class ClusterCanalConnector implements CanalConnector {
currentConnector.connect(); currentConnector.connect();
break; break;
} catch (Exception e) { } catch (Exception e) {
logger.warn("failed to connect to canal server after retry {} times", times); logger.warn("failed to connect to:{} after retry {} times", accessStrategy.currentNode(), times);
currentConnector.disconnect(); currentConnector.disconnect();
currentConnector = null; currentConnector = null;
// retry for #retryTimes for each node when trying to // retry for #retryTimes for each node when trying to

View File

@@ -3,7 +3,7 @@
<parent> <parent>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId> <artifactId>canal</artifactId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<artifactId>canal.common</artifactId> <artifactId>canal.common</artifactId>

View File

@@ -3,7 +3,7 @@
<parent> <parent>
<artifactId>canal.connector</artifactId> <artifactId>canal.connector</artifactId>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
</parent> </parent>
<modelVersion>4.0.0</modelVersion> <modelVersion>4.0.0</modelVersion>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>

View File

@@ -3,7 +3,7 @@
<parent> <parent>
<artifactId>canal.connector</artifactId> <artifactId>canal.connector</artifactId>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>

View File

@@ -4,13 +4,13 @@
<parent> <parent>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId> <artifactId>canal</artifactId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<artifactId>canal.connector</artifactId> <artifactId>canal.connector</artifactId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<packaging>pom</packaging> <packaging>pom</packaging>
<name>canal connector module for otter ${project.version}</name> <name>canal connector module for otter ${project.version}</name>

View File

@@ -3,7 +3,7 @@
<parent> <parent>
<artifactId>canal.connector</artifactId> <artifactId>canal.connector</artifactId>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<modelVersion>4.0.0</modelVersion> <modelVersion>4.0.0</modelVersion>

View File

@@ -3,7 +3,7 @@
<parent> <parent>
<artifactId>canal.connector</artifactId> <artifactId>canal.connector</artifactId>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>

View File

@@ -3,7 +3,7 @@
<parent> <parent>
<artifactId>canal.connector</artifactId> <artifactId>canal.connector</artifactId>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>

View File

@@ -3,7 +3,7 @@
<parent> <parent>
<artifactId>canal.connector</artifactId> <artifactId>canal.connector</artifactId>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>

View File

@@ -3,7 +3,7 @@
<parent> <parent>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId> <artifactId>canal</artifactId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>

View File

@@ -620,30 +620,12 @@ public class QueryLogEvent extends LogEvent {
public static final int Q_DEFAULT_TABLE_ENCRYPTION = 20; public static final int Q_DEFAULT_TABLE_ENCRYPTION = 20;
/** /**
* Replicate PolarDB-X * @since percona 8.0.31 Replicate ddl_skip_rewrite.
*
* @since PolarDB-X 8.0.32
*/
public static final int Q_OPT_FLASHBACK_AREA = 21;
/**
* Replicate PolarDB-X
*
* @since PolarDB-X 8.0.32
*/
public static final int Q_OPT_INDEX_FORMAT_GPP_ENABLED = 22;
/**
* Replicate ddl_skip_rewrite.
*
* @since percona 8.0.31
*/ */
public static final int Q_DDL_SKIP_REWRITE = 21; public static final int Q_DDL_SKIP_REWRITE = 21;
/** /**
* Replicate Q_WSREP_SKIP_READONLY_CHECKS. * @since percona 8.0.31 Replicate Q_WSREP_SKIP_READONLY_CHECKS.
*
* @since percona 8.0.31
*/ */
public static final int Q_WSREP_SKIP_READONLY_CHECKS = 128; public static final int Q_WSREP_SKIP_READONLY_CHECKS = 128;
@@ -661,15 +643,6 @@ public class QueryLogEvent extends LogEvent {
public static final int Q_CHARACTER_SET_COLLATIONS = 131; public static final int Q_CHARACTER_SET_COLLATIONS = 131;
/**
* support PolarDB-X used for storing snapshot tso or commit tso snapshot tso is
* stored in XA End Event with Query_log commit tso is stored in XA Commit Event
* with Query_log
*/
public static final int Q_LIZARD_COMMIT_GCN = 200;
public static final int Q_LIZARD_PREPARE_GCN = 201;
private final void unpackVariables(LogBuffer buffer, final int end) throws IOException { private final void unpackVariables(LogBuffer buffer, final int end) throws IOException {
int code = -1; int code = -1;
try { try {
@@ -767,22 +740,9 @@ public class QueryLogEvent extends LogEvent {
case Q_DEFAULT_TABLE_ENCRYPTION: case Q_DEFAULT_TABLE_ENCRYPTION:
// *start++ = thd->variables.default_table_encryption; // *start++ = thd->variables.default_table_encryption;
buffer.forward(1); buffer.forward(1);
break; case Q_DDL_SKIP_REWRITE:
case Q_OPT_FLASHBACK_AREA:
if (compatiablePercona) {
// percona
// *start++ = thd->variables.binlog_ddl_skip_rewrite; // *start++ = thd->variables.binlog_ddl_skip_rewrite;
buffer.forward(1); buffer.forward(1);
}else {
// PolarDB-X
// *start++ = thd->variables.opt_flashback_area;
buffer.forward(1);
}
break;
case Q_OPT_INDEX_FORMAT_GPP_ENABLED :
// *start++ = thd->variables.opt_index_format_gpp_enabled;
buffer.forward(1);
break;
case Q_HRNOW: case Q_HRNOW:
// https://github.com/alibaba/canal/issues/4940 // https://github.com/alibaba/canal/issues/4940
// percona 和 mariadb各自扩展mysql binlog的格式后有冲突 // percona 和 mariadb各自扩展mysql binlog的格式后有冲突
@@ -819,15 +779,6 @@ public class QueryLogEvent extends LogEvent {
int count = buffer.getUint8(); int count = buffer.getUint8();
// character_set_collations= Lex_cstring((const char *) pos0 , (const char *) pos); // character_set_collations= Lex_cstring((const char *) pos0 , (const char *) pos);
buffer.forward(count * 4); buffer.forward(count * 4);
break;
case Q_LIZARD_COMMIT_GCN:
// commitGCN = buffer.getLong64();
buffer.forward(8);
break;
case Q_LIZARD_PREPARE_GCN:
// prepareGCN = buffer.getLong64();
buffer.forward(8);
break;
default: default:
/* /*
* That's why you must write status vars in growing * That's why you must write status vars in growing
@@ -883,13 +834,9 @@ public class QueryLogEvent extends LogEvent {
return "Q_SQL_REQUIRE_PRIMARY_KEY"; return "Q_SQL_REQUIRE_PRIMARY_KEY";
case Q_DEFAULT_TABLE_ENCRYPTION: case Q_DEFAULT_TABLE_ENCRYPTION:
return "Q_DEFAULT_TABLE_ENCRYPTION"; return "Q_DEFAULT_TABLE_ENCRYPTION";
case Q_OPT_FLASHBACK_AREA: case Q_DDL_SKIP_REWRITE:
// or Q_DDL_SKIP_REWRITE
return "Q_DDL_SKIP_REWRITE";
case Q_OPT_INDEX_FORMAT_GPP_ENABLED:
return "Q_DDL_SKIP_REWRITE"; return "Q_DDL_SKIP_REWRITE";
case Q_HRNOW: case Q_HRNOW:
// or Q_WSREP_SKIP_READONLY_CHECKS
return "Q_HRNOW"; return "Q_HRNOW";
case Q_XID: case Q_XID:
return "Q_XID"; return "Q_XID";

View File

@@ -3,7 +3,7 @@
<parent> <parent>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId> <artifactId>canal</artifactId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>

View File

@@ -17,9 +17,6 @@ canal.instance.master.position=
canal.instance.master.timestamp= canal.instance.master.timestamp=
canal.instance.master.gtid= canal.instance.master.gtid=
# multi stream for polardbx
canal.instance.multi.stream.on=false
# ssl # ssl
#canal.instance.master.sslMode=DISABLED #canal.instance.master.sslMode=DISABLED
#canal.instance.master.tlsVersions= #canal.instance.master.tlsVersions=
@@ -69,4 +66,7 @@ canal.mq.partition=0
#canal.mq.partitionsNum=3 #canal.mq.partitionsNum=3
#canal.mq.dynamicTopicPartitionNum=test.*:4,mycanal:6 #canal.mq.dynamicTopicPartitionNum=test.*:4,mycanal:6
#canal.mq.partitionHash=test.table:id^name,.*\\..* #canal.mq.partitionHash=test.table:id^name,.*\\..*
#
# multi stream for polardbx
canal.instance.multi.stream.on=false
################################################# #################################################

View File

@@ -1,6 +1,6 @@
#!/bin/sh #!/bin/sh
if [ -d "/home/admin/canal-admin" ]; then if [ -d "/home/admin/canal-server" ]; then
serverPort=`perl -le 'print $ENV{"server.port"}'` serverPort=`perl -le 'print $ENV{"server.port"}'`
if [ -z "$serverPort" ] ; then if [ -z "$serverPort" ] ; then
serverPort=8089 serverPort=8089

View File

@@ -11,8 +11,8 @@ SET FOREIGN_KEY_CHECKS = 0;
DROP TABLE IF EXISTS `canal_adapter_config`; DROP TABLE IF EXISTS `canal_adapter_config`;
CREATE TABLE `canal_adapter_config` ( CREATE TABLE `canal_adapter_config` (
`id` bigint(20) NOT NULL AUTO_INCREMENT, `id` bigint(20) NOT NULL AUTO_INCREMENT,
`category` varchar(255) NOT NULL, `category` varchar(45) NOT NULL,
`name` varchar(255) NOT NULL, `name` varchar(45) NOT NULL,
`status` varchar(45) DEFAULT NULL, `status` varchar(45) DEFAULT NULL,
`content` text NOT NULL, `content` text NOT NULL,
`modified_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, `modified_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
@@ -25,7 +25,7 @@ CREATE TABLE `canal_adapter_config` (
DROP TABLE IF EXISTS `canal_cluster`; DROP TABLE IF EXISTS `canal_cluster`;
CREATE TABLE `canal_cluster` ( CREATE TABLE `canal_cluster` (
`id` bigint(20) NOT NULL AUTO_INCREMENT, `id` bigint(20) NOT NULL AUTO_INCREMENT,
`name` varchar(255) NOT NULL, `name` varchar(63) NOT NULL,
`zk_hosts` varchar(255) NOT NULL, `zk_hosts` varchar(255) NOT NULL,
`modified_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, `modified_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`id`) PRIMARY KEY (`id`)
@@ -39,7 +39,7 @@ CREATE TABLE `canal_config` (
`id` bigint(20) NOT NULL AUTO_INCREMENT, `id` bigint(20) NOT NULL AUTO_INCREMENT,
`cluster_id` bigint(20) DEFAULT NULL, `cluster_id` bigint(20) DEFAULT NULL,
`server_id` bigint(20) DEFAULT NULL, `server_id` bigint(20) DEFAULT NULL,
`name` varchar(255) NOT NULL, `name` varchar(45) NOT NULL,
`status` varchar(45) DEFAULT NULL, `status` varchar(45) DEFAULT NULL,
`content` text NOT NULL, `content` text NOT NULL,
`content_md5` varchar(128) NOT NULL, `content_md5` varchar(128) NOT NULL,
@@ -56,7 +56,7 @@ CREATE TABLE `canal_instance_config` (
`id` bigint(20) NOT NULL AUTO_INCREMENT, `id` bigint(20) NOT NULL AUTO_INCREMENT,
`cluster_id` bigint(20) DEFAULT NULL, `cluster_id` bigint(20) DEFAULT NULL,
`server_id` bigint(20) DEFAULT NULL, `server_id` bigint(20) DEFAULT NULL,
`name` varchar(255) NOT NULL, `name` varchar(45) NOT NULL,
`status` varchar(45) DEFAULT NULL, `status` varchar(45) DEFAULT NULL,
`content` text NOT NULL, `content` text NOT NULL,
`content_md5` varchar(128) DEFAULT NULL, `content_md5` varchar(128) DEFAULT NULL,
@@ -72,8 +72,8 @@ DROP TABLE IF EXISTS `canal_node_server`;
CREATE TABLE `canal_node_server` ( CREATE TABLE `canal_node_server` (
`id` bigint(20) NOT NULL AUTO_INCREMENT, `id` bigint(20) NOT NULL AUTO_INCREMENT,
`cluster_id` bigint(20) DEFAULT NULL, `cluster_id` bigint(20) DEFAULT NULL,
`name` varchar(255) NOT NULL, `name` varchar(63) NOT NULL,
`ip` varchar(255) NOT NULL, `ip` varchar(63) NOT NULL,
`admin_port` int(11) DEFAULT NULL, `admin_port` int(11) DEFAULT NULL,
`tcp_port` int(11) DEFAULT NULL, `tcp_port` int(11) DEFAULT NULL,
`metric_port` int(11) DEFAULT NULL, `metric_port` int(11) DEFAULT NULL,
@@ -88,10 +88,10 @@ CREATE TABLE `canal_node_server` (
DROP TABLE IF EXISTS `canal_user`; DROP TABLE IF EXISTS `canal_user`;
CREATE TABLE `canal_user` ( CREATE TABLE `canal_user` (
`id` bigint(20) NOT NULL AUTO_INCREMENT, `id` bigint(20) NOT NULL AUTO_INCREMENT,
`username` varchar(255) NOT NULL, `username` varchar(31) NOT NULL,
`password` varchar(255) NOT NULL, `password` varchar(128) NOT NULL,
`name` varchar(255) NOT NULL, `name` varchar(31) NOT NULL,
`roles` varchar(255) NOT NULL, `roles` varchar(31) NOT NULL,
`introduction` varchar(255) DEFAULT NULL, `introduction` varchar(255) DEFAULT NULL,
`avatar` varchar(255) DEFAULT NULL, `avatar` varchar(255) DEFAULT NULL,
`creation_date` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, `creation_date` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,

View File

@@ -97,6 +97,6 @@ fi
MEMORY="-m 4096m" MEMORY="-m 4096m"
LOCALHOST=`getMyIp` LOCALHOST=`getMyIp`
cmd="docker run -d -it -h $LOCALHOST $CONFIG --name=canal-server $VOLUMNS $NET_MODE $PORTS $MEMORY canal/canal-server" cmd="docker run -d --privileged=true -it -h $LOCALHOST $CONFIG --name=canal-server $VOLUMNS $NET_MODE $PORTS $MEMORY canal/canal-server"
echo $cmd echo $cmd
eval $cmd eval $cmd

View File

@@ -86,6 +86,6 @@ fi
MEMORY="-m 1024m" MEMORY="-m 1024m"
LOCALHOST=`getMyIp` LOCALHOST=`getMyIp`
cmd="docker run -d -it -h $LOCALHOST $CONFIG --name=canal-admin $VOLUMNS $NET_MODE $PORTS $MEMORY canal/canal-admin" cmd="docker run -d --privileged=true -it -h $LOCALHOST $CONFIG --name=canal-admin $VOLUMNS $NET_MODE $PORTS $MEMORY canal/canal-admin"
echo $cmd echo $cmd
eval $cmd eval $cmd

View File

@@ -3,7 +3,7 @@
<parent> <parent>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId> <artifactId>canal</artifactId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>

View File

@@ -196,7 +196,6 @@ public class MysqlConnector {
} }
HandshakeInitializationPacket handshakePacket = new HandshakeInitializationPacket(); HandshakeInitializationPacket handshakePacket = new HandshakeInitializationPacket();
handshakePacket.fromBytes(body); handshakePacket.fromBytes(body);
// default utf8(33)
byte serverCharsetNumber = (handshakePacket.serverCharsetNumber != 0) ? handshakePacket.serverCharsetNumber : 33; byte serverCharsetNumber = (handshakePacket.serverCharsetNumber != 0) ? handshakePacket.serverCharsetNumber : 33;
SslMode sslMode = sslInfo != null ? sslInfo.getSslMode() : SslMode.DISABLED; SslMode sslMode = sslInfo != null ? sslInfo.getSslMode() : SslMode.DISABLED;
if (sslMode != SslMode.DISABLED) { if (sslMode != SslMode.DISABLED) {
@@ -224,10 +223,9 @@ public class MysqlConnector {
serverVersion = handshakePacket.serverVersion; // 记录serverVersion serverVersion = handshakePacket.serverVersion; // 记录serverVersion
logger.info("handshake initialization packet received, prepare the client authentication packet to send"); logger.info("handshake initialization packet received, prepare the client authentication packet to send");
// 某些老协议的 server 默认不返回 auth plugin,需要使用默认的 mysql_native_password // 某些老协议的 server 默认不返回 auth plugin,需要使用默认的 mysql_native_password
String authPluginName = "mysql_native_password"; String authPluginName = (handshakePacket.authPluginName != null
if (handshakePacket.authPluginName != null && handshakePacket.authPluginName.length > 0) { && handshakePacket.authPluginName.length > 0) ? new String(
authPluginName = new String(handshakePacket.authPluginName); handshakePacket.authPluginName) : "mysql_native_password";
}
logger.info("auth plugin: {}", authPluginName); logger.info("auth plugin: {}", authPluginName);
boolean isSha2Password = false; boolean isSha2Password = false;
ClientAuthenticationPacket clientAuth; ClientAuthenticationPacket clientAuth;
@@ -255,7 +253,9 @@ public class MysqlConnector {
logger.info("client authentication packet is sent out."); logger.info("client authentication packet is sent out.");
// check auth result // check auth result
header = null;
header = PacketManager.readHeader(channel, 4); header = PacketManager.readHeader(channel, 4);
body = null;
body = PacketManager.readBytes(channel, header.getPacketBodyLength(), timeout); body = PacketManager.readBytes(channel, header.getPacketBodyLength(), timeout);
assert body != null; assert body != null;
byte marker = body[0]; byte marker = body[0];
@@ -267,7 +267,7 @@ public class MysqlConnector {
body = PacketManager.readBytes(channel, header.getPacketBodyLength(), timeout); body = PacketManager.readBytes(channel, header.getPacketBodyLength(), timeout);
} else { } else {
byte[] authData = null; byte[] authData = null;
String pluginName = authPluginName; String pluginName = null;
if (marker == 1) { if (marker == 1) {
AuthSwitchRequestMoreData packet = new AuthSwitchRequestMoreData(); AuthSwitchRequestMoreData packet = new AuthSwitchRequestMoreData();
packet.fromBytes(body); packet.fromBytes(body);
@@ -294,15 +294,6 @@ public class MysqlConnector {
header = authSwitchAfterAuth(encryptedPassword, header); header = authSwitchAfterAuth(encryptedPassword, header);
body = PacketManager.readBytes(channel, header.getPacketBodyLength(), timeout); body = PacketManager.readBytes(channel, header.getPacketBodyLength(), timeout);
} else if ("caching_sha2_password".equals(pluginName)) { } else if ("caching_sha2_password".equals(pluginName)) {
if (body[0] == 0x01 && body[1] == 0x04) {
// support full auth
// clientAuth提前采用了sha2编码,会减少一次auth交互
header = cachingSha2PasswordFullAuth(channel,
header,
getPassword().getBytes(),
clientAuth.getScrumbleBuff());
body = PacketManager.readBytes(channel, header.getPacketBodyLength(), timeout);
} else {
byte[] scramble = authData; byte[] scramble = authData;
try { try {
encryptedPassword = MySQLPasswordEncrypter.scrambleCachingSha2(getPassword().getBytes(), encryptedPassword = MySQLPasswordEncrypter.scrambleCachingSha2(getPassword().getBytes(),
@@ -317,7 +308,9 @@ public class MysqlConnector {
// fixed issue https://github.com/alibaba/canal/pull/4767, support mysql 8.0.30+ // fixed issue https://github.com/alibaba/canal/pull/4767, support mysql 8.0.30+
header = cachingSha2PasswordFullAuth(channel, header, getPassword().getBytes(), scramble); header = cachingSha2PasswordFullAuth(channel, header, getPassword().getBytes(), scramble);
body = PacketManager.readBytes(channel, header.getPacketBodyLength(), timeout); body = PacketManager.readBytes(channel, header.getPacketBodyLength(), timeout);
} } else {
header = PacketManager.readHeader(channel, 4);
body = PacketManager.readBytes(channel, header.getPacketBodyLength(), timeout);
} }
} else { } else {
header = authSwitchAfterAuth(encryptedPassword, header); header = authSwitchAfterAuth(encryptedPassword, header);
@@ -354,9 +347,9 @@ public class MysqlConnector {
if (packet.status != 0x01) { if (packet.status != 0x01) {
throw new IOException("caching_sha2_password get public key failed"); throw new IOException("caching_sha2_password get public key failed");
} }
logger.info("caching sha2 password fullAuth get server public key succeed."); logger.info("caching sha2 password fullAuth get server public key succeed.");
byte[] publicKeyBytes = packet.authData; byte[] publicKeyBytes = packet.authData;
byte[] encryptedPassword = null; byte[] encryptedPassword = null;
try { try {
encryptedPassword = MySQLPasswordEncrypter.scrambleRsa(publicKeyBytes, pass, seed); encryptedPassword = MySQLPasswordEncrypter.scrambleRsa(publicKeyBytes, pass, seed);

View File

@@ -24,12 +24,9 @@ public class MariaGTIDSet implements GTIDSet {
@Override @Override
public void update(String str) { public void update(String str) {
if (StringUtils.isNotEmpty(str)) {
// 兼容 GTID 为空的情况,例如 mysql-bin.000001 的 GTID 是空的
MariaGtid mariaGtid = MariaGtid.parse(str); MariaGtid mariaGtid = MariaGtid.parse(str);
gtidMap.put(mariaGtid.getDomainId(), mariaGtid); gtidMap.put(mariaGtid.getDomainId(), mariaGtid);
} }
}
public void add(MariaGtid mariaGtid) { public void add(MariaGtid mariaGtid) {
gtidMap.put(mariaGtid.getDomainId(), mariaGtid); gtidMap.put(mariaGtid.getDomainId(), mariaGtid);

View File

@@ -3,30 +3,32 @@ package com.alibaba.otter.canal.parse.driver.mysql.packets.client;
import java.io.ByteArrayOutputStream; import java.io.ByteArrayOutputStream;
import java.io.IOException; import java.io.IOException;
import org.apache.commons.lang.StringUtils;
import com.alibaba.otter.canal.parse.driver.mysql.packets.Capability; import com.alibaba.otter.canal.parse.driver.mysql.packets.Capability;
import com.alibaba.otter.canal.parse.driver.mysql.utils.ByteHelper; import com.alibaba.otter.canal.parse.driver.mysql.utils.ByteHelper;
import com.alibaba.otter.canal.parse.driver.mysql.utils.MSC; import com.alibaba.otter.canal.parse.driver.mysql.utils.MSC;
import com.alibaba.otter.canal.parse.driver.mysql.utils.MySQLPasswordEncrypter; import com.alibaba.otter.canal.parse.driver.mysql.utils.MySQLPasswordEncrypter;
public class ClientAuthenticationSHA2Packet extends ClientAuthenticationPacket { import org.apache.commons.lang.StringUtils;
private int clientCapability = Capability.CLIENT_LONG_PASSWORD | Capability.CLIENT_LONG_FLAG public class ClientAuthenticationSHA2Packet extends ClientAuthenticationPacket {
| Capability.CLIENT_PROTOCOL_41 | Capability.CLIENT_INTERACTIVE
| Capability.CLIENT_TRANSACTIONS | Capability.CLIENT_SECURE_CONNECTION
| Capability.CLIENT_MULTI_STATEMENTS | Capability.CLIENT_PLUGIN_AUTH
| Capability.CLIENT_PLUGIN_AUTH_LENENC_CLIENT_DATA;
@Override @Override
public byte[] toBytes() throws IOException { public byte[] toBytes() throws IOException {
ByteArrayOutputStream out = new ByteArrayOutputStream(); ByteArrayOutputStream out = new ByteArrayOutputStream();
// 1. write client_flags // 1. write client_flags
int capability = clientCapability; int clientCapability = 0;
clientCapability |= Capability.CLIENT_LONG_PASSWORD;
clientCapability |= Capability.CLIENT_LONG_FLAG;
clientCapability |= Capability.CLIENT_PROTOCOL_41;
clientCapability |= Capability.CLIENT_TRANSACTIONS;
clientCapability |= Capability.CLIENT_SECURE_CONNECTION;
clientCapability |= Capability.CLIENT_MULTI_STATEMENTS;
clientCapability |= Capability.CLIENT_PLUGIN_AUTH;
clientCapability |= Capability.CLIENT_PLUGIN_AUTH_LENENC_CLIENT_DATA;
if (getDatabaseName() != null) { if (getDatabaseName() != null) {
capability |= Capability.CLIENT_CONNECT_WITH_DB; clientCapability |= Capability.CLIENT_CONNECT_WITH_DB;
} }
ByteHelper.writeUnsignedIntLittleEndian(capability, out); ByteHelper.writeUnsignedIntLittleEndian(clientCapability, out);
// 2. write max_packet_size // 2. write max_packet_size
ByteHelper.writeUnsignedIntLittleEndian(MSC.MAX_PACKET_LENGTH, out); ByteHelper.writeUnsignedIntLittleEndian(MSC.MAX_PACKET_LENGTH, out);

View File

@@ -3,7 +3,7 @@
<parent> <parent>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId> <artifactId>canal</artifactId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>

View File

@@ -3,7 +3,7 @@
<parent> <parent>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId> <artifactId>canal</artifactId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>

View File

@@ -3,7 +3,7 @@
<parent> <parent>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId> <artifactId>canal</artifactId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../../pom.xml</relativePath> <relativePath>../../pom.xml</relativePath>
</parent> </parent>
<artifactId>canal.instance.core</artifactId> <artifactId>canal.instance.core</artifactId>

View File

@@ -3,7 +3,7 @@
<parent> <parent>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId> <artifactId>canal</artifactId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../../pom.xml</relativePath> <relativePath>../../pom.xml</relativePath>
</parent> </parent>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>

View File

@@ -3,7 +3,7 @@
<parent> <parent>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId> <artifactId>canal</artifactId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>

View File

@@ -3,7 +3,7 @@
<parent> <parent>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId> <artifactId>canal</artifactId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../../pom.xml</relativePath> <relativePath>../../pom.xml</relativePath>
</parent> </parent>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>

View File

@@ -3,7 +3,7 @@
<parent> <parent>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId> <artifactId>canal</artifactId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>

View File

@@ -3,7 +3,7 @@
<parent> <parent>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId> <artifactId>canal</artifactId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<artifactId>canal.parse</artifactId> <artifactId>canal.parse</artifactId>

View File

@@ -46,7 +46,8 @@ import com.taobao.tddl.dbsync.binlog.LogEvent;
public class MysqlEventParser extends AbstractMysqlEventParser implements CanalEventParser, CanalHASwitchable { public class MysqlEventParser extends AbstractMysqlEventParser implements CanalEventParser, CanalHASwitchable {
private CanalHAController haController = null; private CanalHAController haController = null;
private int defaultConnectionTimeoutInSeconds = 30;
private int defaultConnectionTimeoutInSeconds = 30; // sotimeout
private int receiveBufferSize = 64 * 1024; private int receiveBufferSize = 64 * 1024;
private int sendBufferSize = 64 * 1024; private int sendBufferSize = 64 * 1024;
// 数据库信息 // 数据库信息
@@ -68,7 +69,9 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
private int dumpErrorCount = 0; // binlogDump失败异常计数 private int dumpErrorCount = 0; // binlogDump失败异常计数
private int dumpErrorCountThreshold = 2; // binlogDump失败异常计数阀值 private int dumpErrorCountThreshold = 2; // binlogDump失败异常计数阀值
private boolean rdsOssMode = false; private boolean rdsOssMode = false;
private boolean autoResetLatestPosMode = false; // binlog被删除之后,自动按最新的数据订阅 private boolean autoResetLatestPosMode = false; // true:
// binlog被删除之后,自动按最新的数据订阅
private boolean multiStreamEnable;//support for polardbx binlog-x private boolean multiStreamEnable;//support for polardbx binlog-x
protected ErosaConnection buildErosaConnection() { protected ErosaConnection buildErosaConnection() {
@@ -252,6 +255,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
reconnect = true; reconnect = true;
logger.warn("connect failed by ", e); logger.warn("connect failed by ", e);
} }
} }
public MysqlConnection getMysqlConnection() { public MysqlConnection getMysqlConnection() {
@@ -702,10 +706,8 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
*/ */
private EntryPosition findStartPosition(MysqlConnection mysqlConnection) { private EntryPosition findStartPosition(MysqlConnection mysqlConnection) {
try { try {
String showSql = "show binlog events limit 1"; String showSql = multiStreamEnable ?
if (multiStreamEnable) { "show binlog events with " + destination + " limit 1" : "show binlog events limit 1";
showSql = "show binlog events with " + destination + " limit 1";
}
ResultSetPacket packet = mysqlConnection.query(showSql); ResultSetPacket packet = mysqlConnection.query(showSql);
List<String> fields = packet.getFieldValues(); List<String> fields = packet.getFieldValues();
if (CollectionUtils.isEmpty(fields)) { if (CollectionUtils.isEmpty(fields)) {
@@ -726,7 +728,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
@SuppressWarnings("unused") @SuppressWarnings("unused")
private SlaveEntryPosition findSlavePosition(MysqlConnection mysqlConnection) { private SlaveEntryPosition findSlavePosition(MysqlConnection mysqlConnection) {
try { try {
String showSql = "show slave status"; String showSql = "show slave stauts";
if (mysqlConnection.atLeastMySQL84()) { if (mysqlConnection.atLeastMySQL84()) {
// 兼容mysql 8.4 // 兼容mysql 8.4
showSql = "show replica status"; showSql = "show replica status";

View File

@@ -102,7 +102,8 @@ public class DdlResult {
DdlResult ddlResult = this; DdlResult ddlResult = this;
StringBuilder sb = new StringBuilder(); StringBuilder sb = new StringBuilder();
do { do {
sb.append(String.format("DdlResult [schemaName=%s , tableName=%s , oriSchemaName=%s , oriTableName=%s , type=%s ];", sb.append(String.format(
"DdlResult [schemaName=%s , tableName=%s , oriSchemaName=%s , oriTableName=%s , type=%s ];",
ddlResult.schemaName, ddlResult.schemaName,
ddlResult.tableName, ddlResult.tableName,
ddlResult.oriSchemaName, ddlResult.oriSchemaName,

View File

@@ -4,39 +4,18 @@ import java.util.ArrayList;
import java.util.Arrays; import java.util.Arrays;
import java.util.List; import java.util.List;
import org.apache.commons.lang.StringUtils;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.alibaba.polardbx.druid.sql.SQLUtils; import com.alibaba.polardbx.druid.sql.SQLUtils;
import com.alibaba.polardbx.druid.sql.ast.SQLExpr; import com.alibaba.polardbx.druid.sql.ast.SQLExpr;
import com.alibaba.polardbx.druid.sql.ast.SQLStatement; import com.alibaba.polardbx.druid.sql.ast.SQLStatement;
import com.alibaba.polardbx.druid.sql.ast.expr.SQLIdentifierExpr; import com.alibaba.polardbx.druid.sql.ast.expr.SQLIdentifierExpr;
import com.alibaba.polardbx.druid.sql.ast.expr.SQLPropertyExpr; import com.alibaba.polardbx.druid.sql.ast.expr.SQLPropertyExpr;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLAlterTableAddConstraint; import com.alibaba.polardbx.druid.sql.ast.statement.*;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLAlterTableAddIndex;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLAlterTableDropConstraint;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLAlterTableDropIndex;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLAlterTableDropKey;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLAlterTableItem;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLAlterTableRename;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLAlterTableStatement;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLConstraint;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLCreateDatabaseStatement;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLCreateIndexStatement;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLCreateTableStatement;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLDeleteStatement;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLDropDatabaseStatement;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLDropIndexStatement;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLDropTableStatement;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLExprTableSource;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLInsertStatement;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLTableSource;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLTruncateStatement;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLUnique;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLUpdateStatement;
import com.alibaba.polardbx.druid.sql.dialect.mysql.ast.statement.MySqlRenameTableStatement; import com.alibaba.polardbx.druid.sql.dialect.mysql.ast.statement.MySqlRenameTableStatement;
import com.alibaba.polardbx.druid.sql.parser.ParserException; import com.alibaba.polardbx.druid.sql.parser.ParserException;
import com.alibaba.polardbx.druid.util.JdbcConstants; import com.alibaba.polardbx.druid.util.JdbcConstants;
import org.apache.commons.lang.StringUtils;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
/** /**
* @author agapple 2017年7月27日 下午4:05:34 * @author agapple 2017年7月27日 下午4:05:34

View File

@@ -3,26 +3,22 @@ package com.alibaba.otter.canal.parse.inbound.mysql.tsdb;
import java.util.*; import java.util.*;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.parse.inbound.TableMeta;
import com.alibaba.otter.canal.parse.inbound.TableMeta.FieldMeta;
import com.alibaba.otter.canal.parse.inbound.mysql.ddl.DruidDdlParser;
import com.alibaba.otter.canal.protocol.position.EntryPosition;
import com.alibaba.polardbx.druid.DbType; import com.alibaba.polardbx.druid.DbType;
import com.alibaba.polardbx.druid.sql.SQLUtils; import com.alibaba.polardbx.druid.sql.SQLUtils;
import com.alibaba.polardbx.druid.sql.ast.SQLDataType; import com.alibaba.polardbx.druid.sql.ast.SQLDataType;
import com.alibaba.polardbx.druid.sql.ast.SQLDataTypeImpl; import com.alibaba.polardbx.druid.sql.ast.SQLDataTypeImpl;
import com.alibaba.polardbx.druid.sql.ast.SQLExpr; import com.alibaba.polardbx.druid.sql.ast.SQLExpr;
import com.alibaba.polardbx.druid.sql.ast.SQLStatement; import com.alibaba.polardbx.druid.sql.ast.SQLStatement;
import com.alibaba.polardbx.druid.sql.ast.expr.SQLCharExpr; import com.alibaba.polardbx.druid.sql.ast.expr.*;
import com.alibaba.polardbx.druid.sql.ast.expr.SQLIdentifierExpr; import com.alibaba.polardbx.druid.sql.ast.statement.*;
import com.alibaba.polardbx.druid.sql.ast.expr.SQLMethodInvokeExpr;
import com.alibaba.polardbx.druid.sql.ast.expr.SQLNullExpr;
import com.alibaba.polardbx.druid.sql.ast.expr.SQLPropertyExpr;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLColumnConstraint;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLColumnDefinition;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLColumnPrimaryKey;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLColumnUniqueKey;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLCreateTableStatement;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLNotNullConstraint;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLNullConstraint;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLSelectOrderByItem;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLTableElement;
import com.alibaba.polardbx.druid.sql.dialect.mysql.ast.MySqlPrimaryKey; import com.alibaba.polardbx.druid.sql.dialect.mysql.ast.MySqlPrimaryKey;
import com.alibaba.polardbx.druid.sql.dialect.mysql.ast.MySqlUnique; import com.alibaba.polardbx.druid.sql.dialect.mysql.ast.MySqlUnique;
import com.alibaba.polardbx.druid.sql.dialect.mysql.ast.expr.MySqlOrderingExpr; import com.alibaba.polardbx.druid.sql.dialect.mysql.ast.expr.MySqlOrderingExpr;
@@ -32,13 +28,6 @@ import com.alibaba.polardbx.druid.sql.repository.SchemaRepository;
import com.alibaba.polardbx.druid.sql.visitor.SQLASTOutputVisitor; import com.alibaba.polardbx.druid.sql.visitor.SQLASTOutputVisitor;
import com.alibaba.polardbx.druid.sql.visitor.VisitorFeature; import com.alibaba.polardbx.druid.sql.visitor.VisitorFeature;
import com.alibaba.polardbx.druid.util.JdbcConstants; import com.alibaba.polardbx.druid.util.JdbcConstants;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.parse.inbound.TableMeta;
import com.alibaba.otter.canal.parse.inbound.TableMeta.FieldMeta;
import com.alibaba.otter.canal.parse.inbound.mysql.ddl.DruidDdlParser;
import com.alibaba.otter.canal.protocol.position.EntryPosition;
/** /**
* 基于DDL维护的内存表结构 * 基于DDL维护的内存表结构

View File

@@ -4,7 +4,7 @@
<artifactId>canal</artifactId> <artifactId>canal</artifactId>
<packaging>pom</packaging> <packaging>pom</packaging>
<name>canal module for otter ${project.version}</name> <name>canal module for otter ${project.version}</name>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<url>https://github.com/alibaba/canal</url> <url>https://github.com/alibaba/canal</url>
<parent> <parent>
<groupId>org.sonatype.oss</groupId> <groupId>org.sonatype.oss</groupId>
@@ -98,7 +98,7 @@
<java_source_version>1.8</java_source_version> <java_source_version>1.8</java_source_version>
<java_target_version>1.8</java_target_version> <java_target_version>1.8</java_target_version>
<file_encoding>UTF-8</file_encoding> <file_encoding>UTF-8</file_encoding>
<javadoc_skip>false</javadoc_skip> <javadoc_skip>true</javadoc_skip>
<spring_version>5.3.26</spring_version> <spring_version>5.3.26</spring_version>
<log4j_version>2.17.0</log4j_version> <log4j_version>2.17.0</log4j_version>
<rocketmq_version>4.9.8</rocketmq_version> <rocketmq_version>4.9.8</rocketmq_version>

View File

@@ -3,13 +3,13 @@
<parent> <parent>
<artifactId>canal</artifactId> <artifactId>canal</artifactId>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
</parent> </parent>
<modelVersion>4.0.0</modelVersion> <modelVersion>4.0.0</modelVersion>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<artifactId>canal.prometheus</artifactId> <artifactId>canal.prometheus</artifactId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<name>canal prometheus module for otter ${project.version}</name> <name>canal prometheus module for otter ${project.version}</name>
<dependencies> <dependencies>
<dependency> <dependency>

View File

@@ -3,7 +3,7 @@
<parent> <parent>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId> <artifactId>canal</artifactId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>

View File

@@ -3,7 +3,7 @@
<parent> <parent>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId> <artifactId>canal</artifactId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<artifactId>canal.server</artifactId> <artifactId>canal.server</artifactId>

View File

@@ -1,15 +1,5 @@
package com.alibaba.otter.canal.server; package com.alibaba.otter.canal.server;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import com.alibaba.otter.canal.connector.core.config.MQProperties; import com.alibaba.otter.canal.connector.core.config.MQProperties;
import com.alibaba.otter.canal.connector.core.producer.MQDestination; import com.alibaba.otter.canal.connector.core.producer.MQDestination;
import com.alibaba.otter.canal.connector.core.spi.CanalMQProducer; import com.alibaba.otter.canal.connector.core.spi.CanalMQProducer;
@@ -19,6 +9,17 @@ import com.alibaba.otter.canal.instance.core.CanalMQConfig;
import com.alibaba.otter.canal.protocol.ClientIdentity; import com.alibaba.otter.canal.protocol.ClientIdentity;
import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded; import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
public class CanalMQStarter { public class CanalMQStarter {
@@ -106,32 +107,24 @@ public class CanalMQStarter {
public synchronized void startDestination(String destination) { public synchronized void startDestination(String destination) {
CanalInstance canalInstance = canalServer.getCanalInstances().get(destination); CanalInstance canalInstance = canalServer.getCanalInstances().get(destination);
if (canalInstance != null) { if (canalInstance != null) {
CanalMQRunnable canalMQRunnable = canalMQWorks.get(destination); stopDestination(destination);
if (Objects.isNull(canalMQRunnable)) { CanalMQRunnable canalMQRunnable = new CanalMQRunnable(destination);
canalMQRunnable = new CanalMQRunnable(destination);
canalMQWorks.put(canalInstance.getDestination(), canalMQRunnable); canalMQWorks.put(canalInstance.getDestination(), canalMQRunnable);
// 触发一下任务启动 executorService.execute(canalMQRunnable);
Future future = executorService.submit(canalMQRunnable);
canalMQRunnable.setFuture(future);
logger.info("## Start the MQ work of destination:" + destination); logger.info("## Start the MQ work of destination:" + destination);
} else {
// 主段时间内的zk出现session time out,会默认优先当前节点抢占成功
// 如果没有出发过stop动作,这里可以忽略start的启动
logger.info("## Start the MQ work of destination:" + destination + " , ignore stop");
}
} }
} }
public synchronized void stopDestination(String destination) { public synchronized void stopDestination(String destination) {
CanalMQRunnable canalMQRunnable = canalMQWorks.get(destination); CanalMQRunnable canalMQRunable = canalMQWorks.get(destination);
if (canalMQRunnable != null) { if (canalMQRunable != null) {
canalMQRunnable.stop(true); canalMQRunable.stop();
canalMQWorks.remove(destination); canalMQWorks.remove(destination);
logger.info("## Stop the MQ work of destination:" + destination); logger.info("## Stop the MQ work of destination:" + destination);
} }
} }
private void worker(String destination, AtomicBoolean destinationRunning, CountDownLatch latch) { private void worker(String destination, AtomicBoolean destinationRunning) {
while (!running || !destinationRunning.get()) { while (!running || !destinationRunning.get()) {
try { try {
Thread.sleep(100); Thread.sleep(100);
@@ -213,9 +206,6 @@ public class CanalMQStarter {
logger.error("process error!", e); logger.error("process error!", e);
} }
} }
// 确保一下关闭
latch.countDown();
} }
private class CanalMQRunnable implements Runnable { private class CanalMQRunnable implements Runnable {
@@ -228,31 +218,13 @@ public class CanalMQStarter {
private AtomicBoolean running = new AtomicBoolean(true); private AtomicBoolean running = new AtomicBoolean(true);
private CountDownLatch latch = new CountDownLatch(1);
private Future future;
@Override @Override
public void run() { public void run() {
worker(destination, running, latch); worker(destination, running);
} }
public void stop(boolean wait) { public void stop() {
running.set(false); running.set(false);
if (wait) {
try {
// 触发一下interrupt
future.cancel(true);
// 等待MQ发送线程的正常退出
latch.await();
} catch (InterruptedException e) {
// ignore
}
}
}
public void setFuture(Future future) {
this.future = future;
} }
} }
} }

View File

@@ -3,7 +3,7 @@
<parent> <parent>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId> <artifactId>canal</artifactId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>

View File

@@ -3,7 +3,7 @@
<parent> <parent>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId> <artifactId>canal</artifactId>
<version>1.1.8</version> <version>1.1.8-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>