Cluster版本中的Meta

Cluster版本中的Meta

Metadata Client
Metadata Client概述
  1. 定义在 services/meta/client.go中;
  2. Cluster 版本中的Meta是本地的一个内存缓存,数据来源MetaServer;
  3. 对Meta的所有写操作,也将通过http+pb的方式发送到MetaServer, 然后阻塞等待从MetaServer返回的新的Metadata通知;
  4. MetaClient通过http long polling来及时获取Metadata的变化;
所有和Meta data相关的请求

定义在services/meta/internal/meto.proto

enum Type {
        CreateNodeCommand                = 1;
        DeleteNodeCommand                = 2;
        CreateDatabaseCommand            = 3;
        DropDatabaseCommand              = 4;
        CreateRetentionPolicyCommand     = 5;
        DropRetentionPolicyCommand       = 6;
        SetDefaultRetentionPolicyCommand = 7;
        UpdateRetentionPolicyCommand     = 8;
        CreateShardGroupCommand          = 9;
        DeleteShardGroupCommand          = 10;
        CreateContinuousQueryCommand     = 11;
        DropContinuousQueryCommand       = 12;
        CreateUserCommand                = 13;
        DropUserCommand                  = 14;
        UpdateUserCommand                = 15;
        SetPrivilegeCommand              = 16;
        SetDataCommand                   = 17;
        SetAdminPrivilegeCommand         = 18;
        UpdateNodeCommand                = 19;
        CreateSubscriptionCommand        = 21;
        DropSubscriptionCommand          = 22;
        RemovePeerCommand                = 23;
        CreateMetaNodeCommand            = 24;
        CreateDataNodeCommand            = 25;
        UpdateDataNodeCommand            = 26;
        DeleteMetaNodeCommand            = 27;
        DeleteDataNodeCommand            = 28;
        SetMetaNodeCommand               = 29;
    }
重点方法分析
  1. retryUntilExec: 发送请求到MetadataServer, 直到成功返回或到达最大的重试次数;
func (c *Client) retryUntilExec(typ internal.Command_Type, desc *proto.ExtensionDesc, value interface{}) error {
    var err error
    var index uint64
    tries := 0
    currentServer := 0
    var redirectServer string

    for {
        c.mu.RLock()
        // exit if we're closed
        // 如果Client被关闭,我们立即退出
        select {
        case <-c.closing:
            c.mu.RUnlock()
            return nil
        default:
            // we're still open, continue on
        }
        c.mu.RUnlock()

        // build the url to hit the redirect server or the next metaserver
        // 构造请求的Url, 失败时会遍历metaServer发送消息
        var url string
        if redirectServer != "" {
            url = redirectServer
            redirectServer = ""
        } else {
            c.mu.RLock()
            if currentServer >= len(c.metaServers) {
                currentServer = 0
            }
            server := c.metaServers[currentServer]
            c.mu.RUnlock()

            url = fmt.Sprintf("://%s/execute", server)
            if c.tls {
                url = "https" + url
            } else {
                url = "http" + url
            }
        }

        // 发送http请求,成功时返回index,标示当前的metadata版本
        index, err = c.exec(url, typ, desc, value)
        tries++
        currentServer++

        if err == nil {
            // 等待本地的meta data更新到最新, meta data版本用index来标识
            c.waitForIndex(index)
            return nil
        }

        if tries > maxRetries {
            return err
        }

        ...
        
        time.Sleep(errSleep)
    }
}
  1. pollForUpdates: 通过http请求从MetaServer拉取当前MetaData的snapshot,并通知Metadata有改变
    for {
        data := c.retryUntilSnapshot(c.index())
        if data == nil {
            // this will only be nil if the client has been closed,
            // so we can exit out
            return
        }

        // update the data and notify of the change
        c.mu.Lock()
        idx := c.cacheData.Index
        c.cacheData = data
        c.updateAuthCache()
        if idx < data.Index {
            // 通过chan通过Metadata变化 
            close(c.changed)
            c.changed = make(chan struct{})
        }
        c.mu.Unlock()
    }
  1. Client.Open: 从MetaServer拉取meta snapshot并且开启新的goroutine来拉取Metadata更新
func (c *Client) Open() error {
    c.changed = make(chan struct{})
    c.closing = make(chan struct{})
    c.cacheData = c.retryUntilSnapshot(0)

    go c.pollForUpdates()

    return nil
}
Metadata Server
概述
  1. 这是一个CP系统,对metadata采用强一致的存储
  2. raft实现:https://github.com/hashicorp/raft
  3. 存储: https://github.com/hashicorp/raft-boltdb
  4. Meta节点间使用tcp通讯, MetaClient和MetaServer间使用Http通讯
MetaService启动
  1. 定义在services/meta/service.go, Http服务启动
  2. Http请求处理在services/meata/handler.go中, 如果当前的MetaNode不是leader, http请求重定向到Leader,实现上是把leaer http url返回给请求客户端;
Meta请求的执行
  1. Handler.store.apply(body) 来处理具体的请求,走raft一致性写入流程,将序列化后的command作为log写入,log entry被committed后,apply到状态,然后apply返回
  2. Raft相关的操作都定义在service/meta/store.go, 在其open方法初始化raft相关
func (s *store) open(raftln net.Listener) error {
    ...

    var initializePeers []string
    if len(joinPeers) > 0 {
         // 确保其他meta节点的http服务已经open,才继续向下走
        }
    }

    if err := s.setOpen(); err != nil {
        return err
    }

    // Open the raft store.
    // 创建并打开raft store
    if err := s.openRaft(initializePeers, raftln); err != nil {
        return fmt.Errorf("raft: %s", err)
    }

    // 等待leader被选举出来
    if err := s.waitForLeader(0); err != nil {
        return err
    }
    
    ...
    
    return nil
}
  1. command作为log entry被raft给committed后,要apply到fsm, 相应的操作定义在services/meta/store_fsm.go
func (fsm *storeFSM) Apply(l *raft.Log) interface{} {
    var cmd internal.Command
    if err := proto.Unmarshal(l.Data, &cmd); err != nil {
        panic(fmt.Errorf("cannot marshal command: %x", l.Data))
    }

    // Lock the store.
    s := (*store)(fsm)
    s.mu.Lock()
    defer s.mu.Unlock()

    err := func() interface{} {
        switch cmd.GetType() {
        case internal.Command_RemovePeerCommand:
        // 处理各种情况,主要是调用 `services/meta/data.go`中的接口,更改meta信息    
        ...
    }()

    // Copy term and index to new metadata.
    fsm.data.Term = l.Term
    fsm.data.Index = l.Index

    // signal that the data changed
    close(s.dataChanged)
    s.dataChanged = make(chan struct{})

    return err
}
  1. 启动中raft会回调services/meta/store_fsm.go中的Restore接口,从snapshot加载meta信息到store.data
  2. 在上面的Apply函数中,apply成功后,data.index会被更新,同时会调用close(s.dataChanged),通知这个chan作通知
  3. 在上面我们讲过MetaClient通过pollForUpdates来及时取回变更后的MetaData,如果当前MetaData没有变更,即Client和Server端的data.Index是相同的,这个请求将在MetaServer端被hold信;有变更后再返回
func (h *handler) serveSnapshot(w http.ResponseWriter, r *http.Request) {
    ...

    select {
    case <-h.store.afterIndex(index): //等待s.dataChanged的通知,被close后返回
        // Send updated snapshot to client.
        ss, err := h.store.snapshot()
        if err != nil {
            h.httpError(err, w, http.StatusInternalServerError)
            return
        }
        b, err := ss.MarshalBinary()
        if err != nil {
            h.httpError(err, w, http.StatusInternalServerError)
            return
        }
        w.Header().Add("Content-Type", "application/octet-stream")
        w.Write(b)
        return
        
        ...
        
    }
}

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

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏chenssy

【死磕Java并发】----- 死磕 Java 并发精品合集

【死磕 Java 并发】系列是 LZ 在 2017 年写的第一个死磕系列,一直没有做一个合集,这篇博客则是将整个系列做一个概览。

1192
来自专栏向前进

【笔记】HybridApp中使用Promise化的JS-Bridge

背景: HybridApp,前端采用JS-bridge的方式调用Native的接口,如获取设备信息、拍照、人脸识别等 前端封装了调用库,每次调用Native接口...

3484
来自专栏大内老A

了解ASP.NET MVC几种ActionResult的本质:EmptyResult & ContentResult

在之前的两篇文章(《EmptyResult & ContentResult》和《FileResult》)我们剖析了EmptyResult、ContentResu...

2695
来自专栏大内老A

ASP.NET Core应用针对静态文件请求的处理[3]: StaticFileMiddleware中间件如何处理针对文件请求

我们通过《以Web的形式发布静态文件》和《条件请求与区间请求》中的实例演示,以及上面针对条件请求和区间请求的介绍,从提供的功能和特性的角度对这个名为Static...

3295
来自专栏大内老A

通过一个模拟程序让你明白WCF大致的执行流程

在《通过一个模拟程序让你明白ASP.NET MVC是如何运行的》一文中我通过一个普通的ASP.NET Web程序模拟了ASP.NET MVC的执行流程,现在我们...

2136
来自专栏技术博文

mail邮件类

<?php class MailSvc { //-设置全局变量 var $mailTo = ""; // 收件人 var $mailCC = ""; // 抄送...

3719
来自专栏分布式系统进阶

Influxdb的Meta data分析

Influxdb定义了一个Service:Precreator Serivec(services/precreator/service.go),实现比较简单,周...

1302
来自专栏JackieZheng

RabbitMQ入门-Routing直连模式

Hello World模式,告诉我们如何一对一发送和接收消息; Work模式,告诉我们如何多管齐下高效的消费消息; Publish/Subscribe模式,告...

27810
来自专栏偏前端工程师的驿站

asp.net 解码gb2312下urlencode后的字符串

公司网站前期的网页用了gb2312保存用户数据,而我负责的部分用的是utf8,今天恰好要获取前期录入的数据于是毫无悬念地出现乱码问题,经过一番网上的搜索还是找不...

2145
来自专栏雨尘分享

2018 - iOS 面试题汇总一般面试题BAT面试题

4.5K3

扫码关注云+社区

领取腾讯云代金券