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

前言

在上篇文章中，我们讲解了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。