Learn Influxdb the hard way (2) - Dive into the code backbone

本文涉及的产品
容器镜像服务 ACR,镜像仓库100个 不限时长
简介:

前言

在上一篇文章我们从上帝视角鸟瞰了一下Influxdb的组件结构,在这篇文章中,我们会开始深入代码,从代码的流程中帮助大家切割Influxdb的核心组件。希望在看本篇文章的时候,大家能够将Influxdb的代码下载的本地,对照进行查看。

git clone -b v1.5.0 git@github.com:influxdata/influxdb.git

代码主干流程

Influxdb的源码仓库中包含了influx、influx_inspect、influx_stress、influx_tsm、influxd、store等多个子项目,对于本系列而言,更多的侧重在Influxd也就是Influxdb的主要存储Server。

首先我们进入到cmd/influxd/main.go的入口文件,进行代码跟踪,进入到run命令的代码分支流程下。

    //cmd/influxd/run/command.go 133行
    
    s, err := NewServer(config, buildInfo)
    if err != nil {
        return fmt.Errorf("create server: %s", err)
    }
    s.Logger = cmd.Logger
    s.CPUProfile = options.CPUProfile
    s.MemProfile = options.MemProfile
    if err := s.Open(); err != nil {
        return fmt.Errorf("open server: %s", err)
    }
    cmd.Server = s

    // Begin monitoring the server's error channel.
    go cmd.monitorServerErrors()

可以看到Influxdb中Server的构建,并最终调用了Server的Open方法启动Server,这个Server对象是Influxdb的逻辑封装。我们来看下Server包含的内容。

    //cmd/influxd/run/server.go 158行
    
    s.Monitor = monitor.New(s, c.Monitor)
    s.config.registerDiagnostics(s.Monitor)

    if err := s.MetaClient.Open(); err != nil {
        return nil, err
    }

    s.TSDBStore = tsdb.NewStore(c.Data.Dir)
    s.TSDBStore.EngineOptions.Config = c.Data

    // Copy TSDB configuration.
    s.TSDBStore.EngineOptions.EngineVersion = c.Data.Engine
    s.TSDBStore.EngineOptions.IndexVersion = c.Data.Index

    // Create the Subscriber service
    s.Subscriber = subscriber.NewService(c.Subscriber)

    // Initialize points writer.
    s.PointsWriter = coordinator.NewPointsWriter()
    s.PointsWriter.WriteTimeout = time.Duration(c.Coordinator.WriteTimeout)
    s.PointsWriter.TSDBStore = s.TSDBStore

    // Initialize query executor.
    s.QueryExecutor = query.NewQueryExecutor()
    s.QueryExecutor.StatementExecutor = &coordinator.StatementExecutor{
        MetaClient:  s.MetaClient,
        TaskManager: s.QueryExecutor.TaskManager,
        TSDBStore:   coordinator.LocalTSDBStore{Store: s.TSDBStore},
        ShardMapper: &coordinator.LocalShardMapper{
            MetaClient: s.MetaClient,
            TSDBStore:  coordinator.LocalTSDBStore{Store: s.TSDBStore},
        },
        Monitor:           s.Monitor,
        PointsWriter:      s.PointsWriter,
        MaxSelectPointN:   c.Coordinator.MaxSelectPointN,
        MaxSelectSeriesN:  c.Coordinator.MaxSelectSeriesN,
        MaxSelectBucketsN: c.Coordinator.MaxSelectBucketsN,
    }
    s.QueryExecutor.TaskManager.QueryTimeout = time.Duration(c.Coordinator.QueryTimeout)
    s.QueryExecutor.TaskManager.LogQueriesAfter = time.Duration(c.Coordinator.LogQueriesAfter)
    s.QueryExecutor.TaskManager.MaxConcurrentQueries = c.Coordinator.MaxConcurrentQueries

    // Initialize the monitor
    s.Monitor.Version = s.buildInfo.Version
    s.Monitor.Commit = s.buildInfo.Commit
    s.Monitor.Branch = s.buildInfo.Branch
    s.Monitor.BuildTime = s.buildInfo.Time
    s.Monitor.PointsWriter = (*monitorPointsWriter)(s.PointsWriter)

Server的核心代码主要就是实例化内部组件,主要包含Monitor、Subscriber、PointsWriter和QueryExecutor。在上篇文章中已经简单的介绍过这几个组件的作用,在此先不过多赘述,我们再来看下Server的Open方法。

    //cmd/influxd/run/server.go 371行  
    
    s.appendMonitorService()
    s.appendPrecreatorService(s.config.Precreator)
    s.appendSnapshotterService()
    s.appendContinuousQueryService(s.config.ContinuousQuery)
    s.appendHTTPDService(s.config.HTTPD)
    s.appendStorageService(s.config.Storage)
    s.appendRetentionPolicyService(s.config.Retention)
    for _, i := range s.config.GraphiteInputs {
        if err := s.appendGraphiteService(i); err != nil {
            return err
        }
    }
    for _, i := range s.config.CollectdInputs {
        s.appendCollectdService(i)
    }
    for _, i := range s.config.OpenTSDBInputs {
        if err := s.appendOpenTSDBService(i); err != nil {
            return err
        }
    }
    for _, i := range s.config.UDPInputs {
        s.appendUDPService(i)
    }

    s.Subscriber.MetaClient = s.MetaClient
    s.PointsWriter.MetaClient = s.MetaClient
    s.Monitor.MetaClient = s.MetaClient

    s.SnapshotterService.Listener = mux.Listen(snapshotter.MuxHeader)

    // Configure logging for all services and clients.
    if s.config.Meta.LoggingEnabled {
        s.MetaClient.WithLogger(s.Logger)
    }
    s.TSDBStore.WithLogger(s.Logger)
    if s.config.Data.QueryLogEnabled {
        s.QueryExecutor.WithLogger(s.Logger)
    }
    s.PointsWriter.WithLogger(s.Logger)
    s.Subscriber.WithLogger(s.Logger)
    for _, svc := range s.Services {
        svc.WithLogger(s.Logger)
    }
    s.SnapshotterService.WithLogger(s.Logger)
    s.Monitor.WithLogger(s.Logger)

    // Open TSDB store.
    if err := s.TSDBStore.Open(); err != nil {
        return fmt.Errorf("open tsdb store: %s", err)
    }

    // Open the subcriber service
    if err := s.Subscriber.Open(); err != nil {
        return fmt.Errorf("open subscriber: %s", err)
    }

    // Open the points writer service
    if err := s.PointsWriter.Open(); err != nil {
        return fmt.Errorf("open points writer: %s", err)
    }

    s.PointsWriter.AddWriteSubscriber(s.Subscriber.Points())

    for _, service := range s.Services {
        if err := service.Open(); err != nil {
            return fmt.Errorf("open service: %s", err)
        }
    }

在Server的Open中先进行了Service的注册,然后将实例化的内部组件进行启动并设置相应的Logger等配置,最后依次启动注册在Server上的服务。

我们可以挑选appendPrecreatorService作为范例进行分析,来看下一个Service的注册过程。

//cmd/influxd/run/server.go 320行  

func (s *Server) appendPrecreatorService(c precreator.Config) error {
    if !c.Enabled {
        return nil
    }
    srv := precreator.NewService(c)
    srv.MetaClient = s.MetaClient
    s.Services = append(s.Services, srv)
    return nil
}

在本例中讲解的是PrecreatorService这个Service的注册过程,首先实例化一个PrecreatorService,然后设置其使用的MetaClient,最后将PrecreatorService的实例注册到Server上,在Influxdb中Service是一个接口类型,需要实现如下三个方法,分别负责统一的日志接入,启动和停止。

//cmd/influxd/run/server.go 558行

// Service represents a service attached to the server.
type Service interface {
    WithLogger(log *zap.Logger)
    Open() error
    Close() error
}

这样Influxdb的代码基本已经切割清晰了,首先是一个Server负责实例化几个全局的内部组件单例,然后生成上层业务包装的Service,最后再依次启动。

最后

在接下来的文章中,我们会深入到每个Service中,依次讲解他们的功能与原理。如果有错误的地方麻烦大家多多指正。

目录
相关文章
|
8月前
|
自然语言处理 算法 知识图谱
DEGREE: A Data-Efficient Generation-Based Event Extraction Model论文解读
事件抽取需要专家进行高质量的人工标注,这通常很昂贵。因此,学习一个仅用少数标记示例就能训练的数据高效事件抽取模型已成为一个至关重要的挑战。
73 0
|
10月前
|
TensorFlow 算法框架/工具 Python
tensorflow安装错误:Could not find a version that satisfies the requirement tensorflow 解决
tensorflow安装错误:Could not find a version that satisfies the requirement tensorflow 解决
|
机器学习/深度学习 算法 数据挖掘
【多标签文本分类】Improved Neural Network-based Multi-label Classification with Better Initialization ……
【多标签文本分类】Improved Neural Network-based Multi-label Classification with Better Initialization ……
【多标签文本分类】Improved Neural Network-based Multi-label Classification with Better Initialization ……
sbs
|
存储 SQL 人工智能
The Volcano Optimizer Generator: Extensibility and Efficient Search 论文翻译
原文:The Volcano Optimizer Generator: Extensibility and Efficient SearchThe Volcano Optimizer Generator: Extensibility and Efficient Search 论文翻译。2023.01.25 —— by zz【中括号内为译者注】对原文部分关键术语,或重点句有加粗。便于定位。为了避免英
sbs
182 0
The Volcano Optimizer Generator: Extensibility and Efficient Search 论文翻译
《40 Must Know Questions to test a data scientist on Dimensionality Reduction techniques》电子版地址
40 Must Know Questions to test a data scientist on Dimensionality Reduction techniques
75 0
《40 Must Know Questions to test a data scientist on Dimensionality Reduction techniques》电子版地址
|
Python
《Data Pre-Processing in PythonHow I learned to love parallelized applies with Dask and Numba》电子版地址
Data Pre-Processing in Python:How I learned to love parallelized applies with Dask and Numba
64 0
《Data Pre-Processing in PythonHow I learned to love parallelized applies with Dask and Numba》电子版地址
|
TensorFlow 算法框架/工具
|
TensorFlow 算法框架/工具
《A beginner introduction to TensorFlow (Part-1)》电子版地址
A beginner introduction to TensorFlow (Part-1)
80 0
《A beginner introduction to TensorFlow (Part-1)》电子版地址
|
Shell 计算机视觉
2022亚太建模A题Feature Extraction of Sequence Images and Modeling Analysis of Mold Flux Melting and Crystallization思路分析
2022 亚太建模A题序列图像的特征提取与建模分析 模具流量的熔融和结晶Feature Extraction of Sequence Images and Modeling Analysis of Mold Flux Melting and Crystallization
2022亚太建模A题Feature Extraction of Sequence Images and Modeling Analysis of Mold Flux Melting and Crystallization思路分析
《Investigation of Transformer based Spelling Correction Model for CTC-based End-to-End Mandarin Speech Recognition》电子版地址
Investigation of Transformer based Spelling Correction Model for CTC-based End-to-End Mandarin Speech Recognition
75 0
《Investigation of Transformer based Spelling Correction Model for CTC-based End-to-End Mandarin Speech Recognition》电子版地址