LearnInfluxdbthehardway(5)ServicesinInfluxdbIII

前言

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

SnapshotterService 解析

首先我们来看下SnapshotterService的定义

// services/snapshotter/service.go 31行
// Service manages the listener for the snapshot endpoint.
type Service struct {wg sync.WaitGroupNode *influxdb.NodeMetaClient interface {encoding.BinaryMarshalerDatabase(name string) *meta.DatabaseInfo}TSDBStore interface {BackupShard(id uint64, since time.Time, w io.Writer) errorExportShard(id uint64, ExportStart time.Time, ExportEnd time.Time, w io.Writer) errorShard(id uint64) *tsdb.ShardShardRelativePath(id uint64) (string, error)SetShardEnabled(shardID uint64, enabled bool) errorRestoreShard(id uint64, r io.Reader) errorCreateShard(database, retentionPolicy string, shardID uint64, enabled bool) error}Listener net.ListenerLogger *zap.Logger
}

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

// services/snapshotter/service.go 124行if RequestType(typ[0]) &＃61;&＃61; RequestShardUpdate {return s.updateShardsLive(conn)}r, bytes, err :&＃61; s.readRequest(conn)if err !&＃61; nil {return fmt.Errorf("read request: %s", err)}switch RequestType(typ[0]) {case RequestShardBackup:if err :&＃61; s.TSDBStore.BackupShard(r.ShardID, r.Since, conn); err !&＃61; nil {return err}case RequestShardExport:if err :&＃61; s.TSDBStore.ExportShard(r.ShardID, r.ExportStart, r.ExportEnd, conn); err !&＃61; nil {return err}case RequestMetastoreBackup:if err :&＃61; s.writeMetaStore(conn); err !&＃61; 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中&＃xff0c;会根据RequestType的不同&＃xff0c;进行不同行为的处理&＃xff0c;其中RequestShardUpdate、RequestShardBackup、RequestShardExport是只和Shard有关的操作&＃xff0c;因此直接透传给了TSDBStore。而RequestMetastoreBackup、RequestDatabaseInfo、RequestRetentionPolicyInfo涉及到Influxdb中数据库的一些信息查询&＃xff0c;因此在Snapshooter中通过调用MetaClient进行了封装。

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

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

//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&＃61;%s since %s", cmd.retentionPolicy, cmd.since.Format(time.RFC3339))} else {cmd.StdoutLogger.Printf("backing up rp&＃61;%s with boundaries start&＃61;%s, end&＃61;%s",cmd.retentionPolicy, cmd.start.Format(time.RFC3339), cmd.end.Format(time.RFC3339))}req :&＃61; &snapshotter.Request{Type: snapshotter.RequestRetentionPolicyInfo,BackupDatabase: cmd.database,BackupRetentionPolicy: cmd.retentionPolicy,}response, err :&＃61; cmd.requestInfo(req)if err !&＃61; nil {return err}return cmd.backupResponsePaths(response)
}

最后

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