Influxdb的Meta data分析

Meta分析

一图抵千言

metadata.png

按上图,我们从右到从逐一简单介绍一下
ShardInfo
  1. 定义了一个Shard的id和它位于哪个data node上;
ShardGroupInfo
  1. 封装了ShardGroup的相关信息
  2. Influxdb是按时间写入数据的,每个DB都有自己的Retention Policy(这个我们后面会介绍),这个Retention Policy规定了每两个ShardGroup之间的时间跨度ShardGroup Duration, 即每过一个ShardGrup Duration就会生产切换到下一个新的ShardGroup;
  3. 这个ShardGroupInfo就记录了当前这个ShardGroup的相关信息,比较主要的信息有: 3.1 StartTime: 这个Group里最早的时间 3.2 EndTime: 这个Group里最晚的时间 3.3 根据上面的两个时间,我们就可以按时间和时间范围来查找到相应的ShardGroup; 3.4 Shards []ShardInfo: 这个ShardGroup包含的所有Shard,对于同一个ShardGroup,按Series key(Point key)不同散列写到不同的Shard中;
RetentionPolicyInfo
  1. 封装了Retention Policy: 包括了复本个数,数据保留时长,ShardGroup切分时长和当前节点的所有ShardGroup信息
  2. 定义了按时间和时间范围查找相应SahrdGroup的方法
DatabaseInfo
  1. 管理 RetentionPoliciesContinuousQueries
UserInfo
  1. 封装了用户信息:用户名,密码,对db的操作权限
总结
  1. 上面介绍的每个对象基础都提供了对其管理的下层metadata信息的增,删,查的方法;

Meta Client

定义
  1. 定义在services/meta/client.go中,负责所有和meta data有关的操作和请求处理
type Client struct {
    logger *zap.Logger

    mu        sync.RWMutex
    closing   chan struct{}
    changed   chan struct{}
    cacheData *Data

    // Authentication cache.
    authCache map[string]authUser

    path string

    retentionAutoCreate bool
}

主要就是操作上面介绍过的cacheData *Data;

  1. 提供了大量的方法,基出上都是对上述Data类型包含的meta信息的增,删,查,改操作
主要方法介绍
  1. snapshot方法:将meta数据写入磁盘,所有的meta信息都有对应的protocol buffer结构,依赖protocol buffer作序列化和反序列化:
func snapshot(path string, data *Data) error {
    filename := filepath.Join(path, metaFile)
    tmpFile := filename + "tmp"

    f, err := os.Create(tmpFile)

    defer f.Close()

    var d []byte
    //利用protocol buffer作二进制的序列化
    if b, err := data.MarshalBinary(); err != nil {
        return err
    } else {
        d = b
    }

    //写入文件 
    if _, err := f.Write(d); err != nil {
        return err
    }

    if err = f.Sync(); err != nil {
        return err
    }

    //close file handle before renaming to support Windows
    if err = f.Close(); err != nil {
        return err
    }

    return file.RenameFile(tmpFile, filename)
}
  1. Load方法:meta数据是会保存到磁盘的,influxdb启动时也会从磁盘上读取:
func (c *Client) Load() error {
    file := filepath.Join(c.path, metaFile)

    f, err := os.Open(file)
    defer f.Close()

    data, err := ioutil.ReadAll(f)

    //利用protocol buffer作反序列化
    if err := c.cacheData.UnmarshalBinary(data); err != nil {
        return err
    }
    return nil
}
  1. commit方法:influxdb运行时,所有的meta信息在内存里都缓存一分,当meta信息有改动时,通过此方法立即写入磁盘,同时更新内存里的缓存
func (c *Client) commit(data *Data) error {
    data.Index++

    // try to write to disk before updating in memory
    if err := snapshot(c.path, data); err != nil {
        return err
    }

    // update in memory
    c.cacheData = data

    // close channels to signal changes
    close(c.changed)
    c.changed = make(chan struct{})

    return nil
}
  1. ShardGroupsByTimeRangeShardsByTimeRange:按给定的时间查找已有的ShardGroup和Shard
func (c *Client) ShardGroupsByTimeRange(database, policy string, min, max time.Time) (a []ShardGroupInfo, err error) {
    ...
    // 先找到RetentionPolicyInfo
    rpi, err := c.cacheData.RetentionPolicy(database, policy)
    if err != nil {
        return nil, err
    } else if rpi == nil {
        return nil, influxdb.ErrRetentionPolicyNotFound(policy)
    }
    groups := make([]ShardGroupInfo, 0, len(rpi.ShardGroups))
    
    //遍历RPI中的所有ShardGroup
    for _, g := range rpi.ShardGroups {
        if g.Deleted() || !g.Overlaps(min, max) {
            continue
        }
        groups = append(groups, g)
    }
    return groups, nil
}
  1. PrecreateShardGroups: 预先创建ShardGroup, 避免在相应时间段数据到达时才创建ShardGroup
func (c *Client) PrecreateShardGroups(from, to time.Time) error {
    c.mu.Lock()
    defer c.mu.Unlock()
    data := c.cacheData.Clone()
    var changed bool

    // 遍历所有的DatabaseInfo信息
    for _, di := range data.Databases {
        for _, rp := range di.RetentionPolicies {
            if len(rp.ShardGroups) == 0 {
                // No data was ever written to this group, or all groups have been deleted.
                continue
            }
            
            // ShardGroups中的所有ShardGroup已经是按时间排序好的,最后一个也就是最新的一个ShardGroup
            g := rp.ShardGroups[len(rp.ShardGroups)-1] // Get the last group in time.
            
            // 
            if !g.Deleted() && g.EndTime.Before(to) && g.EndTime.After(from) {
                // Group is not deleted, will end before the future time, but is still yet to expire.
                // This last check is important, so the system doesn't create shards groups wholly
                // in the past.

                // Create successive shard group.
                // 计算出需要创建的ShardGroup的开始时间
                nextShardGroupTime := g.EndTime.Add(1 * time.Nanosecond)
                // if it already exists, continue
                if sg, _ := data.ShardGroupByTimestamp(di.Name, rp.Name, nextShardGroupTime); sg != nil {
                    continue
                }
                newGroup, err := createShardGroup(data, di.Name, rp.Name, nextShardGroupTime)
                if err != nil {
                    continue
                }
                changed = true
            }
        }
    }

    if changed {
        if err := c.commit(data); err != nil {
            return err
        }
    }

    return nil
}

Influxdb定义了一个Service:Precreator Serivec(services/precreator/service.go),实现比较简单,周期性的调用PrecreateShardGroups,看是否需要创建ShardGroup

func (s *Service) runPrecreation() {
    defer s.wg.Done()

    for {
        select {
        case <-time.After(s.checkInterval):
            if err := s.precreate(time.Now().UTC()); err != nil {
                s.Logger.Info("Failed to precreate shards", zap.Error(err))
            }
        case <-s.done:
            s.Logger.Info("Terminating precreation service")
            return
        }
    }
}

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏智能大石头

实体处理模块IEntityModule

在2015年7月16日,XCode新增了实体处理模块IEntityModule,用于拦截实体对象添删改操作。

1160
来自专栏JavaEdge

高性能队列——Disruptor总论1 背景2 Java内置队列3 ArrayBlockingQueue的问题4 Disruptor的设计方案代码样例性能等待策略Log4j 2应用场景

这里所说的队列是系统内部的内存队列,而不是Kafka这样的分布式队列 Disruptor特性限于3.3.4

2113
来自专栏草根专栏

从头编写 asp.net core 2.0 web api 基础框架 (4) EF配置

Github源码地址:https://github.com/solenovex/Building-asp.net-core-2-web-api-starter-...

7477
来自专栏Linux驱动

46.Linux-分析rc红外遥控平台驱动框架,修改内核的NEC解码函数BUG(1)

1.2然后在drivers\media\rc\keymaps里存了各种不同的键映射文件

2262
来自专栏智能大石头

实体处理模块IEntityModule

在2015年7月16日,XCode新增了实体处理模块IEntityModule,用于拦截实体对象添删改操作。 该接口参考IHttpModule设计理念,横切在实...

19710
来自专栏大内老A

ASP.NET MVC涉及到的5个同步与异步,你是否傻傻分不清楚?[上篇]

Action方法的执行具有两种基本的形式,即同步执行和异步执行,而在ASP.NETMVC的整个体系中涉及到很多同步/异步的执行方式,虽然在前面相应的文章中已经对...

2256
来自专栏老九学堂

【面试】找工作必看的十道XML面试题

XML并不依赖于其他编程语言,与SQL一样是编程人员所必备的技能之一,因此在任何技术工作面试之前准备一些XML问题都是很有意义的。老九君为大家整合了十道有关XM...

4457
来自专栏你不就像风一样

史上超全面的Elasticsearch使用指南

elasticsearch简写es,es是一个高扩展、开源的全文检索和分析引擎,它可以准实时地快速存储、搜索、分析海量的数据。

2.5K2
来自专栏好好学java的技术栈

java后端开发每天遇到的jsp,了解一下

JSP脚本片断(scriptlet)用于在JSP页面中编写多行Java代码(在<%%>不能定义方法)。语法:<%多行java代码 %>

2272
来自专栏java 成神之路

RocketMQ 底层通信机制 源码分析

RocketMQ 底层通讯是使用Netty来实现的。 下面我们通过源码分析下RocketMQ是怎么利用Netty进行通讯的。

1522

扫码关注云+社区

领取腾讯云代金券