Learn Influxdb the hard way (4) - Services in Influxdb II-阿里云开发者社区

开发者社区> 阿里云容器服务 ACK> 正文

Learn Influxdb the hard way (4) - Services in Influxdb II

简介:

前言

在上篇文章中我们讲解了Influxdb中MonitorService的工作原理,在本篇文章中会继续讲解PrecreatorService,从名字上来看,这个Service的作用是预先创建一些资源,那么具体Influxdb会预先初始化哪些资源,以及这些资源原先创建的优势是什么呢,让我们今天一起来从代码里面找答案。

PrecreatorService解析

首先我们来看下PrecreatorService的创建过程,从这个Service的结构中我们可以看到它包含了两个时间的属性checkInterval和advancePeriod,此外包含了一个MetaClient接口的引用以及一个waitgroup,由此可以猜测,这个Service的主要实现方式是通过周期性的通过多个并发的goroutines实现创建资源,然后再讲meta信息通过MetaClient的实例应用写入到meta.db中。

//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
}

// services/precreator/service.go 12行
// Service manages the shard precreation service.
type Service struct {
    checkInterval time.Duration
    advancePeriod time.Duration

    Logger *zap.Logger

    done chan struct{}
    wg   sync.WaitGroup

    MetaClient interface {
        PrecreateShardGroups(now, cutoff time.Time) error
    }
}

// NewService returns an instance of the precreation service.
func NewService(c Config) *Service {
    return &Service{
        checkInterval: time.Duration(c.CheckInterval),
        advancePeriod: time.Duration(c.AdvancePeriod),
        Logger:        zap.NewNop(),
    }
}

下面我们就通过实际的这个Service的运行Code进行下验证。从源码中可以发现,这个Service主要作用就是在checkInterval的周期到达的时候,通过时间戳进行ShardGroups的创建。

// services/precreator/service.go 71行
// runPrecreation continually checks if resources need precreation.
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
        }
    }
}

// precreate performs actual resource precreation.
func (s *Service) precreate(now time.Time) error {
    cutoff := now.Add(s.advancePeriod).UTC()
    return s.MetaClient.PrecreateShardGroups(now, cutoff)
}

在代码中有一个很有趣的用法cutoff := now.Add(s.advancePeriod).UTC(),这个的作用是什么呢,查询了一下官方的文档,这个参数的主要作用是提前多长时间创建ShardGroups,默认情况下checkInterval的取值为10m,而advancePeriod的取值为30m,也就是说Influxdb会每10分钟进行一下检查,看是否需要为30分钟后预创建ShardGroups。接下来深入到创建ShardGroups的源码中。

// services/meta/client.go 777行
// PrecreateShardGroups creates shard groups whose endtime is before the 'to' time passed in, but
// is yet to expire before 'from'. This is to avoid the need for these shards to be created when data
// for the corresponding time range arrives. Shard creation involves Raft consensus, and precreation
// avoids taking the hit at write-time.
func (c *Client) PrecreateShardGroups(from, to time.Time) error {
    c.mu.Lock()
    defer c.mu.Unlock()
    data := c.cacheData.Clone()
    var changed bool

    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
            }
            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.
                nextShardGroupTime := g.EndTime.Add(1 * time.Nanosecond)
                // if it already exists, continue
                if sg, _ := data.ShardGroupByTimestamp(di.Name, rp.Name, nextShardGroupTime); sg != nil {
                    c.logger.Info("Shard group already exists",
                        logger.ShardGroup(sg.ID),
                        logger.Database(di.Name),
                        logger.RetentionPolicy(rp.Name))
                    continue
                }
                newGroup, err := createShardGroup(data, di.Name, rp.Name, nextShardGroupTime)
                if err != nil {
                    c.logger.Info("Failed to precreate successive shard group",
                        zap.Uint64("group_id", g.ID), zap.Error(err))
                    continue
                }
                changed = true
                c.logger.Info("New shard group successfully precreated",
                    logger.ShardGroup(newGroup.ID),
                    logger.Database(di.Name),
                    logger.RetentionPolicy(rp.Name))
            }
        }
    }

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

    return nil
}

在前几篇文章中我们稍微涉及了一些关于Influxdb存储结构上的内容,一个Influxdb可以包含多个库,每个库的存储是根据retention policy来分目录的,而retention policy之下才是真正的Shard,因此在预创建Shard的时候,需要为每一个符合条件的库以及库之下的retention policy都创建Shard。
如果此时retention policy下没有任何的Shard,则会认为无需预创建,因为大部分的情况下,这种场景意味着数据被清空的中间态。为了保证创建的Shard的时序性,在预创建的时候会获取一个retention policy下的最后一个shard,并检查当前shard的所属时间序列是否会和新创建的Shard有重合,只有新创建的Shard与最新的Shard之间不存在重合关系的时候,才会进行创建。继续跟踪代码,查看创建Shard的流程。

// services/meta/data.go 350行
// CreateShardGroup creates a shard group on a database and policy for a given timestamp.
func (data *Data) CreateShardGroup(database, policy string, timestamp time.Time) error {
    // Find retention policy.
    rpi, err := data.RetentionPolicy(database, policy)
    if err != nil {
        return err
    } else if rpi == nil {
        return influxdb.ErrRetentionPolicyNotFound(policy)
    }

    // Verify that shard group doesn't already exist for this timestamp.
    if rpi.ShardGroupByTimestamp(timestamp) != nil {
        return nil
    }

    // Create the shard group.
    data.MaxShardGroupID++
    sgi := ShardGroupInfo{}
    sgi.ID = data.MaxShardGroupID
    sgi.StartTime = timestamp.Truncate(rpi.ShardGroupDuration).UTC()
    sgi.EndTime = sgi.StartTime.Add(rpi.ShardGroupDuration).UTC()
    if sgi.EndTime.After(time.Unix(0, models.MaxNanoTime)) {
        // Shard group range is [start, end) so add one to the max time.
        sgi.EndTime = time.Unix(0, models.MaxNanoTime+1)
    }

    data.MaxShardID++
    sgi.Shards = []ShardInfo{
        {ID: data.MaxShardID},
    }

    // Retention policy has a new shard group, so update the policy. Shard
    // Groups must be stored in sorted order, as other parts of the system
    // assume this to be the case.
    rpi.ShardGroups = append(rpi.ShardGroups, sgi)
    sort.Sort(ShardGroupInfos(rpi.ShardGroups))

    return nil
}

当我们看到创建Shard的代码逻辑的时候,其实是有些失望的,此时创建的Shard支持在内存中进行了创建,并没有在磁盘中创建相应的目录结构,那么在磁盘上面Shard的目录结构和对应的Shard的名字是怎么来的呢。实际上这个Shard的信息只是在meta与内存中作了一个标示,并不是真的开始存储数据,这部分的内容会在PointsWriter.WritePoints的时候刷到磁盘上的,在后面解析PointsWriter的部分会进行深入的讨论。

总结

至此PrecreatorService的部分已经比较明晰了,他的原理非常简单,通过异步定时执行,建立在内存中和meta数据中的shard索引,为后续写入数据的Shard分布提供索引,这样当数据通过PointsWriter写入数据时会根据对应的索引hash分布到真正的Shard中。也就是说通过预分配Shard的部分可以让数据更好的Hash到不同的存储文件中,通过时间戳可以对应到Shard,从而可以使得数据更好的切分,更快的查询。

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

分享:
阿里云容器服务 ACK
使用钉钉扫一扫加入圈子
+ 订阅

云端最佳容器应用运行环境,安全、稳定、极致弹性

官方博客
官网链接