Influxdb的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
}

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

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
}

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
}

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
}

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
}

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