分布式对象存储 读书笔记(三) 元数据

上节的代码无法区分同一个对象的不同版本，为了记录对象版本以及其他一些元数据，本节中会加入一个新组件：元数据服务

元数据服务就是提供对元数据的存取功能的服务。元数据指的是对象的描述信息，比如对象的名字、版本、大小以及散列值.

新增加服务后的结构图如下

需要新安装一个ElasticSearch搜索引擎。安装过程可自行百度

ElasticSearch搜索引擎中，我们需要定义这样一个映射。

'{"mappings":{"objects":{"properties":{"name":{"type":"string","index":"not_analyzed"},"version":{"type":"integer"},"size":{"type":"integer"},"hash":{"type":"string"}}}}}'

ES的索引相当于数据库而类型相当于数据库的表，映射则相当于定义表结构。该索引只有一个类型就是objects,包括4个属性分别是name,version,size和hash

接口服务代码添加了versions访问处理函数

func main() {
    go heartbeat.ListenHeartbeat()
    http.HandleFunc("/objects/", objects.Handler)
    http.HandleFunc("/locate/", locate.Handler)
    http.HandleFunc("/versions/", versions.Handler)
    log.Fatal(http.ListenAndServe(os.Getenv("LISTEN_ADDRESS"), nil))
}

version处理函数仅处理Get请求

调用ES的API进行version查询最后返回

func Handler(w http.ResponseWriter, r *http.Request) {
    m := r.Method
    if m != http.MethodGet {
        w.WriteHeader(http.StatusMethodNotAllowed)
        return
    }
    from := 0
    size := 1000
    name := strings.Split(r.URL.EscapedPath(), "/")[2]
    for {
        metas, e := es.SearchAllVersions(name, from, size)
        if e != nil {
            log.Println(e)
            w.WriteHeader(http.StatusInternalServerError)
            return
        }
        for i := range metas {
            b, _ := json.Marshal(metas[i])
            w.Write(b)
            w.Write([]byte("
"))
        }
        if len(metas) != size {
            return
        }
        from += size
    }
}

OBJECT处理函数中 添加了DELETE处理

在ES中添加新版本，数据为0 ，哈希为"",表示删除操作

OBJECT处理函数中 PUT操作没有较大改变 但是添加了在ES中记录元数据的代码

OBJECT处理函数中 GET操作会根据version查询ES 获取元数据 才开始获取对象数据 

func get(w http.ResponseWriter, r *http.Request) {
    name := strings.Split(r.URL.EscapedPath(), "/")[2]
    versionId := r.URL.Query()["version"]
    version := 0
    var e error
    if len(versionId) != 0 {
        version, e = strconv.Atoi(versionId[0])
        if e != nil {
            log.Println(e)
            w.WriteHeader(http.StatusBadRequest)
            return
        }
    }
    meta, e := es.GetMetadata(name, version)
    if e != nil {
        log.Println(e)
        w.WriteHeader(http.StatusInternalServerError)
        return
    }
    if meta.Hash == "" {
        w.WriteHeader(http.StatusNotFound)
        return
    }
    object := url.PathEscape(meta.Hash)
    stream, e := getStream(object)
    if e != nil {
        log.Println(e)
        w.WriteHeader(http.StatusNotFound)
        return
    }
    io.Copy(w, stream)
}

func put(w http.ResponseWriter, r *http.Request) {
    hash := utils.GetHashFromHeader(r.Header)
    if hash == "" {
        log.Println("missing object hash in digest header")
        w.WriteHeader(http.StatusBadRequest)
        return
    }

    c, e := storeObject(r.Body, url.PathEscape(hash))
    if e != nil {
        log.Println(e)
        w.WriteHeader(c)
        return
    }
    if c != http.StatusOK {
        w.WriteHeader(c)
        return
    }

    name := strings.Split(r.URL.EscapedPath(), "/")[2]
    size := utils.GetSizeFromHeader(r.Header)
    e = es.AddVersion(name, hash, size)
    if e != nil {
        log.Println(e)
        w.WriteHeader(http.StatusInternalServerError)
    }
}

实操验证

开启 rabbitmq ES 设置ES映射

开启 apiserver dataserver

使用curl 上传一个test3对象  这时候会显示400 bad request。 因为没提供test3 对象的散列值 

先计算对象的哈希值的 然后再添加哈希PUT一次

OK！！