Canal 是阿里巴巴集团提供的一个开源产品，能够通过解析数据库的增量日志，提供增量数据的订阅和消费功能。使用 Canal 模拟成 MySQL 的 Slave，实时接收 MySQL 的增量数据 binlog，然后通过 RESTful API 将数据写入到 Easysearch 中。

前提条件

1. 部署 Easysearch 集群。
2. 部署 MySQL 数据库。
3. 部署 Gateway，Canal Adapter 不支持使用 HTTPS 协议连接，使用 Gateway 代理 Easysearch 。
4. 部署 Console，方便查看 Easysearch 数据。
   对于自建 MySQL , 需要先开启 Binlog 写入功能，配置 binlog-format 为 ROW 模式，my.cnf 中配置如下：

   [mysqld]
   log-bin=mysql-bin # 开启 binlog
   binlog-format=ROW # 选择 ROW 模式
   server_id=1 # 配置 MySQL replaction 需要定义，不要和 canal 的 slaveId 重复

   创建 canal 用户，授权 canal 连接 MySQL 具有作为 MySQL slave 的权限。

   CREATE USER canal IDENTIFIED BY 'canal';
   GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
   -- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;
   FLUSH PRIVILEGES;

   操作步骤

   在进行数据同步时支持自定义索引 Mapping，但需保证 Mapping 中定义的字段（名称+类型）与 MySQL 中一致。

   1. 准备 MySQL 数据源

   create database canal；
   use canal;
   CREATE TABLE `test` (
       `id` bigint(32) NOT NULL,
       `name` text NOT NULL,
       `age` smallint  NOT NULL,
       PRIMARY KEY (`id`)
   ) ENGINE=InnoDB
   DEFAULT CHARACTER SET=utf8;

   2. Easysearch 创建索引

   PUT test
   {
       "settings" : {
         "index" : {
           "number_of_shards" : "1",
           "number_of_replicas" : "1"
         }
       },
       "mappings" : {
               "properties" : {
                 "id": {
                      "type": "integer"
                  },
                  "name": {
                       "type" : "text"
                   },
                   "age" : {
                       "type" : "integer"
                   }
               }
       }
   }

   3. 安装并启动 Canal-server

   下载：https://github.com/alibaba/canal/releases/download/canal-1.1.7/canal.deployer-1.1.7.tar.gz  
   修改配置文件 vi conf/example/instance.properties
   启动 canal  
   sh bin/startup.sh  
   启动成功日志信息，logs/canal/canal.log 关闭 canal  
   sh bin/stop.sh

   4. 安装并启动 Canal-adapter

   下载：https://github.com/alibaba/canal/releases/download/canal-1.1.7/canal.adapter-1.1.7.tar.gz  
   修改配置文件：application.yml

   server:
     port: 8081
   spring:
     jackson:
       date-format: yyyy-MM-dd HH:mm:ss
       time-zone: GMT+8
       default-property-inclusion: non_null
   canal.conf:
     flatMessage: true
     syncBatchSize: 1000
     retries: -1
     timeout:
     accessKey:
     secretKey:
     consumerProperties:
       canal.tcp.server.host: 127.0.0.1:11111
       canal.tcp.batch.size: 500
     srcDataSources:
       defaultDS:
         url: jdbc:mysql://127.0.0.1:3306/canal?useUnicode=true
         username: canal
         password: canal
     canalAdapters:
       groups:
       - groupId: g1
         outerAdapters:
         - name: logger
         - name: es7
           properties:
             security.auth: admin:4ad8f8f792e81cd0a6de
             cluster.name: easysearch

   新增 canal-adapter/conf/es7/test.yml，配置索引和表的映射关系。

   dataSourceKey: defaultDS
   destination: example
   groupId: g1
   esMapping:
     _index: test           # es 的索引名称
     _id: _id               # es 的_id, 如果不配置该项必须配置下面的pk项_id则会由es自动分配
     # sql映射
     sql: " select a.id as _id,a.id,a.name,a.age from test a "
     etlCondition: "where a.id>={}"
     commitBatch: 3000      # 提交批大小

   启动 canal-adapter  
   ./bin/startup.sh
   

   5. 验证增量数据同步

   在 MySQL 数据库中，对 test 表插入两条数据。  
   inserttest(id,name,age) values(1,'canal_test1',11);  
inserttest(id,name,age) values(2,'canal_test2',22);

   6. 在 Console 中，执行以下命令查询数据

   

   最后

   Canal 同步的是增量数据，不会同步之前的存量数据。要同步存量数据可参考《使用 Logstash 同步 MySQL 到 Easysearch》

INFINI Labs 产品又更新啦~，本次更新概要如下：发布 Easysearch-client Java 客户端，开发者通过 client 与 Easysearch 集群的交互变得更加简洁和直观；Console 开发工具新增 SQL 特性，支持 SELECT 查询等语法高亮和自动提示等；Gateway 的系统 API 添加了基于基本身份验证的安全功能。

以下是本次更新的详细说明。

INFINI Easysearch-client v1.0.1

正式发布 Easysearch Java 客户端。

这一里程碑式的更新为开发人员带来了前所未有的便利性，使得与 Easysearch 集群的交互变得更加简洁和直观。现在，通过 Easysearch-client 客户端，开发者可以直接使用 Java 方法和数据结构来进行交互，而不再需要依赖于传统的 HTTP 方法和 JSON。这一变化大大简化了操作流程，使得数据管理和索引更加高效。高级客户端的功能范围包括处理数据操作，管理集群，包括查看和维护集群的健康状态，并对 Security 模块全面兼容。它提供了一系列 API，用于管理角色、用户、权限、角色映射和账户。这意味着安全性和访问控制现在可以更加细粒度地管理，确保了数据的安全性和合规性。

使用说明参见：快速开始

INFINI Console v1.11.0

INFINI Console 是一款非常轻量级的多集群、跨版本的搜索基础设施统一管控平台。通过对流行的搜索引擎基础设施进行跨版本、多集群的集中纳管， 企业可以快速方便的统一管理企业内部的不同版本的多套搜索集群。

Console 在线体验： http://demo.infini.cloud (用户名/密码：readonly/readonly)。

Console 本次更新如下：

Features

• 开发工具 SQL 查询支持
  • 支持 SELECT 查询及语法高亮
  • 支持索引和字段自动提示
  • 支持 FROM 前置语法

Bug fix

• 修复平台概览集群指标为空的问题

Improvements

• LDAP 支持从 DN 中解析 OU 属性
• 集群动态优化显示，新增节点名称和索引名称的聚合统计过滤

INFINI Gateway v1.19.0

INFINI Gateway 是一个面向搜索场景的高性能数据网关，所有请求都经过网关处理后再转发到后端的搜索业务集群。基于 INFINI Gateway 可以实现索引级别的限速限流、常见查询的缓存加速、查询请求的审计、查询结果的动态修改等等。

Gateway 本次更新如下：

Features

• 添加 http 处理器
• 在 API 模块中添加基于基本身份验证的安全性
• 允许将自身注册到配置管理器
• 允许在配置错误时触发 panic

Bug fix

• 修复 rewrite_to_bulk 在较新版本中缺少 _type 的问题
• 修复 rewrite_to_bulk，支持无索引文档操作

Improvements

• 更新测试，断言解析结果

期待反馈

欢迎下载体验使用，如果您在使用过程中遇到如何疑问或者问题，欢迎前往 INFINI Labs Github（https://github.com/infinilabs） 中的对应项目中提交 Feature Request 或提交 Bug。

• 下载地址： https://www.infinilabs.com/download

您还可以通过邮件联系我们：hello@infini.ltd

或者拨打我们的热线电话：(+86) 400-139-9200

欢迎加入 Discord 聊天室：https://discord.gg/4tKTMkkvVX

也欢迎大家微信扫码添加小助手（INFINI-Labs），加入用户群一起讨论交流。

关于极限科技（INFINI Labs）

极限科技，全称极限数据（北京）科技有限公司，是一家专注于实时搜索与数据分析的软件公司。旗下品牌极限实验室（INFINI Labs）致力于打造极致易用的数据探索与分析体验。

极限科技是一支年轻的团队，采用天然分布式的方式来进行远程协作，员工分布在全球各地，希望通过努力成为中国乃至全球企业大数据实时搜索分析产品的首选，为中国技术品牌输出添砖加瓦。

官网：https://www.infinilabs.com

近年来，日志管理平台越来越流行。使用日志管理平台可以实时地、统一地、方便地管理和查看日志，挖掘日志数据价值，驱动运维、运营，提升服务管理效率。

方案架构

• Beats 是轻量级采集器，包括 Filebeat、Metricbeat 等。
• Easysearch 是个分布式搜索引擎，提供搜集、分析、存储数据等主要功能。
• Console 是一个可视化工具，提供可视化查询，制作报表等功能。

本文将搭建一个统一日志管理平台。使用 Filebeat 采集 OS 中的日志（其他日志大同小异），发送到 Easysearch 中。最后通过 Console 进行日志的可视化查询与分析。

操作步骤

1. 准备工作
   • 部署 Easysearch
     • 编辑 easysearch.yml 文件，打开注释 elasticsearch.api_compatibility: true
   • 部署 Console
2. 安装并配置 Filebeat
   • 下载 Filebeat，版本建议选择 7.10.2。下载地址：https://www.elastic.co/cn/downloads/past-releases/filebeat-oss-7-10-2
   • 修改配置 filebeat.yml 文件

setup.template.name: "filebeat"
setup.template.pattern: "system-log*"
setup.template.fields: "${path.config}/fields.yml"

output.elasticsearch:
    hosts: ["localhost:9200"]
    protocol: "https"
    ssl.verification_mode: none
    username: "admin"
    password: "4ad8f8f792e81cd0a6de"
    index: "system-log"

3. 启用 system 模块并导入 pipeline

./filebeat modules enable system
./filebeat setup --pipelines --modules system

4. 创建索引模板及初始索引，使用 ZSTD+SOURCE_REUSE 技术节省磁盘空间

PUT _template/system_log
{
    "order": 100,
  "index_patterns": [
      "system_log*"
    ],
      "settings": {
        "index": {
          "format": "7",
          "lifecycle": {
          "name": "ilm_.infini_metrics-30days-retention",
          "rollover_alias": "system_log"
        },
        "codec": "ZSTD",
        "source_reuse": true,
        "number_of_shards": "1",
        "translog": {
          "durability": "async"
        }
      }
    },
    "mappings": {
      "dynamic_templates": [
        {
          "strings": {
            "mapping": {
              "ignore_above": 256,
              "type": "keyword"
            },
            "match_mapping_type": "string"
          }
        }
      ]
    }
}

PUT system-log-00001
{
    "aliases":{
    "system-log":{
      "is_write_index":true
    }
  }
}

5. 启动 filebeat

nohup ./filebeat -c filebeat.yml 2>&1>/dev/null &

6. 进入 Console 查看、搜索日志
   
7. 进入 Console 创建 dashboard 进行日志分析