Learn Influxdb the hard way (5) - Services in Influxdb III

本文涉及的产品
容器镜像服务 ACR,镜像仓库100个 不限时长
容器服务 Serverless 版 ACK Serverless,317元额度 多规格
容器服务 Serverless 版 ACK Serverless,952元额度 多规格
简介:

前言

在上篇文章中,我们讲解了Influxdb中的PrecreatorService的工作原理,在这篇文章中,我们会将讲解SnapshotterService,从这个Service的名字来看是和数据备份相关的,接下来继续进入到Code中。

SnapshotterService 解析

首先我们来看下SnapshotterService的定义

// services/snapshotter/service.go 31行
// Service manages the listener for the snapshot endpoint.
type Service struct {
    wg sync.WaitGroup

    Node *influxdb.Node

    MetaClient interface {
        encoding.BinaryMarshaler
        Database(name string) *meta.DatabaseInfo
    }

    TSDBStore interface {
        BackupShard(id uint64, since time.Time, w io.Writer) error
        ExportShard(id uint64, ExportStart time.Time, ExportEnd time.Time, w io.Writer) error
        Shard(id uint64) *tsdb.Shard
        ShardRelativePath(id uint64) (string, error)
        SetShardEnabled(shardID uint64, enabled bool) error
        RestoreShard(id uint64, r io.Reader) error
        CreateShard(database, retentionPolicy string, shardID uint64, enabled bool) error
    }

    Listener net.Listener
    Logger   *zap.Logger
}

从结构上来看,这个Service的原理已经差不多可以看懂了,MetaClient主要用来查询和db相关的meta,然后针对meta中存储的shard相关信息去操作TSDBStore,然后进行Shard维度的备份恢复操作,此外这个Service还包含了一个Listener,也就是说这个Service会启动一个4层的Server来对外提供调用服务。顺着这个思路,去向下翻看serve的Code。

// services/snapshotter/service.go 124行
    
    if RequestType(typ[0]) == RequestShardUpdate {
        return s.updateShardsLive(conn)
    }

    r, bytes, err := s.readRequest(conn)
    if err != nil {
        return fmt.Errorf("read request: %s", err)
    }

    switch RequestType(typ[0]) {
    case RequestShardBackup:
        if err := s.TSDBStore.BackupShard(r.ShardID, r.Since, conn); err != nil {
            return err
        }
    case RequestShardExport:
        if err := s.TSDBStore.ExportShard(r.ShardID, r.ExportStart, r.ExportEnd, conn); err != nil {
            return err
        }
    case RequestMetastoreBackup:
        if err := s.writeMetaStore(conn); err != nil {
            return err
        }
    case RequestDatabaseInfo:
        return s.writeDatabaseInfo(conn, r.BackupDatabase)
    case RequestRetentionPolicyInfo:
        return s.writeRetentionPolicyInfo(conn, r.BackupDatabase, r.BackupRetentionPolicy)
    case RequestMetaStoreUpdate:
        return s.updateMetaStore(conn, bytes, r.BackupDatabase, r.RestoreDatabase, r.BackupRetentionPolicy, r.RestoreRetentionPolicy)
    default:
        return fmt.Errorf("request type unknown: %v", r.Type)
    }

在Serve中,会根据RequestType的不同,进行不同行为的处理,其中RequestShardUpdate、RequestShardBackup、RequestShardExport是只和Shard有关的操作,因此直接透传给了TSDBStore。而RequestMetastoreBackup、RequestDatabaseInfo、RequestRetentionPolicyInfo涉及到Influxdb中数据库的一些信息查询,因此在Snapshooter中通过调用MetaClient进行了封装。

到此为止,其实这个Service已经不需要过多的赘述其中的实现了,但是问题在于这样的Service看上去很不友好,很难被上层的系统系统调用,在Influxdb中是怎么使用这个Service呢。

回到代码的目录中,果然我们发现了一个client.go的文件,看上去为了调用Snapshotter相关的功能Influxdb还提供了一个客户端SDK,最后在Influxd的backup命令中,印证了我们的分析。

//cmd/influxd/backup/backup.go 342行
// backupRetentionPolicy will request the retention policy information from the server and then backup
// every shard in the retention policy. Each shard will be written to a separate file.
func (cmd *Command) backupRetentionPolicy() error {
    if cmd.isBackup {
        cmd.StdoutLogger.Printf("backing up rp=%s since %s", cmd.retentionPolicy, cmd.since.Format(time.RFC3339))
    } else {
        cmd.StdoutLogger.Printf("backing up rp=%s with boundaries start=%s, end=%s",
            cmd.retentionPolicy, cmd.start.Format(time.RFC3339), cmd.end.Format(time.RFC3339))
    }

    req := &snapshotter.Request{
        Type:                  snapshotter.RequestRetentionPolicyInfo,
        BackupDatabase:        cmd.database,
        BackupRetentionPolicy: cmd.retentionPolicy,
    }

    response, err := cmd.requestInfo(req)
    if err != nil {
        return err
    }

    return cmd.backupResponsePaths(response)
}

最后

我们再回顾下SnapshotterService,这个Service是负责数据的备份和恢复的,涉及Influxdb数据库级别的部分会通过MetaClient进行操作,涉及Shard的具体存储部分,会透传给TSDBStore,最后建立了一个4层的Server,并提供上层的SDK给Influxd的命令进行调用。在下一篇文章中,我们会分析最重要的HTTPDService。

目录
相关文章
Query Performance Optimization at Alibaba Cloud Log Analytics Service
PrestoCon Day 2023,链接:https://prestoconday2023.sched.com/event/1Mjdc?iframe=no首页自我介绍,分享题目概要各个性能优化项能够优化的资源类别limit快速短路有什么优点?有啥特征?进一步的优化空间?避免不必要块的生成逻辑单元分布式执行,global 阶段的算子哪些字段无需输出?公共子表达式结合FilterNode和Proje
Query Performance Optimization at Alibaba Cloud Log Analytics Service
|
机器学习/深度学习 Ubuntu iOS开发
【Elastic Engineering】Beats:解密 Filebeat 中的 setup 命令
这个步骤非常重要,但是描述的内容并不是很多。为什么需要这个步骤呢?它到底能够做什么呢?
574 0
【Elastic Engineering】Beats:解密 Filebeat 中的 setup 命令
|
存储 API 索引
【Elastic Engineering】Elasticsearch:Cluster 备份 Snapshot 及 Restore API
Elasticsearch:Cluster 备份 Snapshot 及 Restore API
265 0
【Elastic Engineering】Elasticsearch:Cluster 备份 Snapshot 及 Restore API
|
存储 测试技术 API
【Elastic Engineering】Elasticsearch:Runtime fields 入门, Elastic 的 schema on read 实现 - 7.11 发布
Elasticsearch:Runtime fields 入门, Elastic 的 schema on read 实现 - 7.11 发布
211 0
【Elastic Engineering】Elasticsearch:Runtime fields 入门, Elastic 的 schema on read 实现 - 7.11 发布
|
Web App开发 存储 人工智能
|
存储 API 索引
|
索引
|
存储 Java 开发者
|
5G API 索引
|
数据可视化 API 索引
【Elastic Engineering】Elasticsearch:创建 Runtime field 并在 Kibana 中使用它 - 7.11 发布
Elasticsearch:创建 Runtime field 并在 Kibana 中使用它 - 7.11 发布
359 0
【Elastic Engineering】Elasticsearch:创建 Runtime field 并在 Kibana 中使用它 - 7.11 发布