传统应用

一般情况下，你要对外提供 HTTP 接口，首先你得开发一个 Web Server，然后部署到服务器上，直接对外提供 HTTP 接口。
后来，你发现流量高了，你又没时间优化，只能堆机器了，于是乎你横向扩展了一堆 Web Server，这个时候你就需要一个 Proxy Server 做反向代理（Reverse Proxy）来让你的 Web Server 统一对外提供 HTTP 接口，并且尽可能的保证这一堆 Web Server 承载的流量是一样一样的。

微服务

随着时间的推移，你的 Web Server 不断壮大，承载的功能也越来越多了，这个时候微服务（Microservices）出现在你眼中，确认过眼神，你选择拆分你的 Web Server 的功能模块。为了更好的性能，你决定不再使用 HTTP 作为内部通信的方式了，RPC 有幸被你选中了。于是你重新设计了整个架构图。

那么问题来了，你的 RPC Server 提供的可能不是传统的 HTTP 协议了，这样搞，你就得找一个支持 RPC 反代的 Proxy 了，幸运的是 Nginx 支持了 gRPC 的反向代理，但是浏览器不一定能正常解析你的 RPC 的报文，RPC 一般情况下需要特定的 Client 去支持。所以这个时候，你就需要在 Proxy 与 RPC Server 之间加一个网关（Gateway）去做协议转换，将 RPC 的与 HTTP 之间做转换，让 Prxoy 之前的服务不需要做任何改变，就可以支持目前的架构。

Gateway

协议转换

参数重构

Gateway 对外提供传统的 HTTP 接口，对内，可以作为 RPC Client 去调用对应的 RPC Server。并且 Gateway 还要能够自动发现你新加入的 RPC Server，这样在紧急扩容，或者新接入了 RPC Server，可以不间断，自动的让 RPC Server 暴露出去。

首先来说一下协议转换，一个 HTTP 请求进来

POST /posts?namespace=huilu HTTP/1.1
Host: hui.lu
User-Agent: curl/7.54.0
Accept: */*
X-Token: f136803ab9c241079ba0cc1b5d02ee77
Cookie: UserID=1
Content-Type: application/json; charset=utf-8
Content-Length: 14

{"foo": "bar"}

上面报文中的任何信息，都可以是一个参数，并且可能是后端 RPC Server 需要的，但是 RPC Server 接受的参数形式又不是上述 HTTP 报文中的格式，所以这里就需要 Gateway 来做一下转换。

我们这里假设后端是 gRPC，那么它的这个 POST 请求对应的 protobuf 如下

service Project {
  rpc CreateProject (PostsRequest) returns (PostsResponse) {}
}

message PostsRequest {
  string namespace = 1;
  int64 user_id = 2;
  map<string, string> params = 3;
  string token = 3;
}

message PostsResponse {
  int64 project_id = 1;
}

从上面来看，HTTP 请求的参数与 gRPC 参数的对应关系如下

上面就能很清晰的看出来 HTTP 请求的参数与 gRPC 调用的参数映射关系，这里就需要设计一个好用的配置语法，来描述参数的映射关系。在这里，我们选择 YAML 来作为配置的语法

url: /posts
method: post
request:
    gRPC.PostsRequest.user_id: HTTP.Header.Cookie.UserID
    gRPC.PostsRequest.namespace: HTTP.Query.namespace
    gRPC.PostsRequest.token: HTTP.Header.X-Token
    gRPC.PostsRequest.params: HTTP.Body

上面的配置，比上面的表格就强多了，人肉眼可见，理解起来也方便多了， 而且给代码来识别解析，也是很方便。但是上面有一些冗余的地方，比如 request 里面的 key，都是以 gRPC.PostsRequest 开始，后面 value 也都是以 HTTP 开始，在这个特定的场景下，完全可以省略掉

url: /posts
method: post
request:
    user_id: Header.Cookie.UserID
    namespace: Query.namespace
    token: Header.X-Token
    params: Body

可是，假设 gRPC 请求参数中多了一个，需要一个常量 type，并且值恒定为 Body，并且是特定值，不想要 HTTP 请求去改变它，那么上述的配置文件就变成了这样的

url: /posts
method: post
request:
    user_id: Header.Cookie.UserID
    namespace: Query.namespace
    token: Header.X-Token
    params: Body
    type: Body

可是，上面的配置文件就多出了两个 Body 的 value 了，这样会有歧义，你并不能明确的说明，Body 到底是 HTTP 请求的 body 还是一个字符串。首先你想到的是，程序在解析 gRPC 的 protobuf 时，分析出来 gRPC.PostsRequest. type 是个 string 类型，所以请求的 type 也必须是个 string 类型，因为 HTTP.Body 在这个场景下，肯定不是 string，所以程序可以自动处理掉这种情景，不会将 HTTP.Body 当做 type 的值去解析，但是当 HTTP.Body 也是个 string 咋整。所以，我们这里将 HTTP 请求的参数加一个特定的，短小精悍的标识符 $

url: /posts
method: post
request:
    user_id: $.Header.Cookie.UserID
    namespace: $.Query.namespace
    token: $.Header.X-Token
    params: $.Body
    type: Body

上面是不是就明确多了，我们假设 $. 开头的表达式都是我们的参数表达式，并且会从对应的 HTTP 报文中找到对应的值，在调用 gRPC 时，传给后端。
请求说完了，再来说一下响应，上述的 gRPC 返回的是一个 PostsResponse 的对象，我们假设原来的 HTTP 接口返回的也是一样的结构，但是参数名不一样，HTTP 响应的数据是

{
  "id": 1,
  "type": "Body"
}

那么在 gRPC 的响应与 HTTP 的响应数据之间也要做一次参数的转换，还是遵循上面的配置规则

url: /posts
method: post
request:
    user_id: $.Header.Cookie.UserID
    namespace: $.Query.namespace
    token: $.Header.X-Token
    params: $.Body
    type: Body
response:
    id: gRPC.PostsResponse.project_id
    type: Body

这里，gRPC.PostsResponse.project_id 在这里还是有点多余，我们也可以将其替换成 $.rpc.project_id，我们在这里，规定 $ 的生命周期是从请求进入到返回响应都是存在的，来作为一个 rootScope，里面保存了请求和响应的各种参数，那么就简化成了下面的形式

request:
    user_id: $.Header.Cookie.UserID
    namespace: $.Query.namespace
    token: $.Header.X-Token
    params: $.Body
    type: Body
response:
    id: $.rpc.project_id
    type: Body

Gateway 就可以利用上面的配置，转换 HTTP 接口与 gRPC 调用的参数，我们在这里，将上述的参数转换，叫做「参数重构」。

插件机制

当你做完上面的事情后，你把你的 Gateway 卖给了前端，RPC Server 的业务方，这个时候，他们就来说了，这里的很多请求，都需要鉴定 token 是否有效，如果每次参数都要加上 token 多累啊，后端也只需要一个 user_id 而已，于是你灵机一动，既然大家都有这种重复逻辑的代码，那我就把它抽出来，在 Gateway 这里做这件事，大家只有在配置里加一个选项，就可以了。
于是，你就加了一个 auth 的配置，当 auth 为 true 的时候，你就调用一个专门用来做认证的 RPC Server，认证后，就会返回给你用户信息，你就搞出了一个 $.auth.user_id 的配置，只要大家在配置里写了这个，就能获取到 user_id

url: /posts
method: post
auth: true
request:
    user_id: $.auth.user_id
    namespace: $.Query.namespace
    params: $.Body
    type: Body
response:
    id: $.rpc.project_id
    type: Body

这个时候，大家就都拍手叫好，不停的说「666」，可是这个时候，又有个人跳出来说，他不止需要 user_id，还需要 mobile 等信息，然后你就又在 $.auth 上加了一个 $.auth.mobile，这样会又获取到了 mobile 了。当你正在准备看看监控喝茶的时候，公司推出了认证服务二代，这个时候，就不是验证 $.Header.Cookie.Token 了，你就想，不就是再加个 auth2 吗？

url: /posts
method: post
auth2: true
request:
    user_id: $.auth.user_id
    namespace: $.Query.namespace
    params: $.Body
    type: Body
response:
    id: $.rpc.project_id
    type: Body

但是，当你写出这种配置的时候，你发现，这看起来太 XX 丑了，而且后面加了认证三代、四代咋整，总不能直接加载配置里面吧，于是你也开始拆分 Gateway 的逻辑，将参数重构作为核心功能，将认证相关的拆分成一个个插件，于是配置文件就变成了下面这样的了

url: /posts
method: post
plugins:
    - auth
request:
    user_id: $.auth.user_id
    namespace: $.Query.namespace
    params: $.Body
    type: Body
response:
    id: $.rpc.project_id
    type: Body

当你正在为自己的设计沾沾自喜的时候，有人来告诉你，他需要用到两种认证方式，这个时候，你就发现了自己埋下的坑了，你将两种认证服务的用户信息都放在了 $.auth 上了，gg，没法区分了，你就想，要不加一个 $.auth2 来保存认证二代的信息？当你发现你的脑袋里出现了这种念头时，你自己都觉得可怕，于是你就想，为啥不能在插件使用的时候，指定用户信息保存的地方呢？

url: /posts
method: post
plugins:
    - auth --output $.auth
    - auth-v2 --output $.auth2
request:
    user_id: $.auth.user_id
    namespace: $.Query.namespace
    params: $.Body
    type: Body
response:
    id: $.rpc.project_id
    type: Body

于是，你又一次满足了客户的要求，客户连连称赞你。但是有个人过来告诉你，由于历史原因，认证用的 token 没有放在 HTTP.Header.X-Token，而是放在了 HTTP.Header.Cookie.TOKEN，你当成懵逼，但是你突然想到你刚加了一个 --output，那为啥不能加一个 --input 来指定 token 的地方了？

url: /posts
method: post
plugins:
    - auth --output $.auth --input $.Cookie.TOKEN
    - auth-v2 --output $.auth2
request:
    user_id: $.auth.user_id
    namespace: $.Query.namespace
    params: $.Body
    type: Body
response:
    id: $.rpc.project_id
    type: Body

你不止改进了插件，顺带给常用的 Cookie 做了一个别名，直接可以用过 $.Cookie 来取 Cookie 的值。

执行时机

后续你又加了一系列的插件，比如字符串替换 string-replace，可以设置某个值的 set 插件。但是这些插件，你都是作用在请求 request 解析之前的，你突然想到，要是有人要求对 response 的某个字段做字符串替换，那不就 gg 了吗，于是你决定将这种问题扼杀在摇篮中。
你仔细研究了下上面的配置，和协议转换的流程，你发现在这里可以划分为多个阶段

• 请求参数重构前
• 请求参数解析后，RPC 调用前
• RPC 调用后，响应参数重构前
• 响应参数重构后，返回响应前

上面描述废话有点多，我们这里简化下

• 请求参数重构前（beforeRequestParamsRefactor）
• RPC 调用前（beforeRPCCall）
• 响应参数重构前（beforeRPCResponseParamsRefactor）
• 返回响应前（beforeRPCResponse）

在这里我们划分为四个阶段，我们用 set 作为示例

url: /posts
method: post
plugins:
    - auth --output $.auth --input $.Cookie.TOKEN
    - auth-v2 --output $.auth2
    - set $.request.type Body --timing beforeRPCCall
    - set $.response.type Body --timing beforeRPCResponse
request:
    user_id: $.auth.user_id
    namespace: $.Query.namespace
    params: $.Body
response:
    id: $.rpc.project_id

我们规定，所有的插件默认执行时机都是 beforeRequestParamsRefactor，所以上面我们用 --timing 去改变了插件的执行时机。利用 set 插件，将常量 Body 直接强制覆盖掉请求和响应中的 type。

内部路由

但是上述的配置，只能调用一个 RPC Server，有一个接口，他需要同时调用两个 RPC Server，并且拆分请求参数和合并响应结果，于是你就想，为什么不能有一个没有 url 的配置

name: get_something
request:
    id: $.request.project_id
response:
    data: $.rpc.data

于是就有了上面的命名路由，这里我们叫它内部路由，因为它没法被外部直接调用，那这个应该怎么与上述的结合在一起呢？我们将内部路由定义成 $.route.get_something

url: /posts
method: post
plugins:
    - auth --output $.auth --input $.Cookie.TOKEN
    - auth-v2 --output $.auth2
    - set $.request.type Body --timing beforeRPCCall
    - set $.response.type Body --timing beforeRPCResponse
request:
    user_id: $.auth.user_id
    namespace: $.Query.namespace
    params: $.Body
response:
    id: $.rpc.project_id
    something: $.route.get_something

服务发现

上述编故事，忘了插入 RPC 调用了 :D，其实也是没地方能植入 RPC 调用的地方，于是就将这个地方抽出来，单独讲。
假设一个 RPC Server，你有 100 个实例来提供服务

• 192.168.1.2:8100
• 192.168.1.2:8102
• 192.168.1.3:8104
• ...

这么多机器，你怎么知道机器在哪里呢，就几台机器，你可以手动写到配置里

url: /posts
method: post
request:
    user_id: $.auth.user_id
    namespace: $.Query.namespace
    params: $.Body
response:
    id: $.rpc.project_id
rpc:
    - host: 192.168.1.2
      port: 8100
    - host: 192.168.1.2
      port: 8102
    - host: 192.168.1.3
      port: 8104

可是，现在 Docker 盛行，RPC Server 的 IP 是会不停的改变的，所以怎么能保证实例的 IP 能够每次自动的让 Gateway 知道呢？这个时候，就需要服务发现了，首先每个 RPC Server 在启动的时候，需要将自己的 IP:PORT 告诉到一个地方，注册中心，然后也要告诉注册中心，它是谁，也就是它的名字，这样，Gateway 就可以通过 RPC Server 的名字，到注册中心从，查到它的地址了，并且在地址改变了之后，注册中心也会通过各种方式通知 Gateway，去新的地方调它。

url: /posts
method: post
request:
    user_id: $.auth.user_id
    namespace: $.Query.namespace
    params: $.Body
response:
    id: $.rpc.project_id
rpc:
    name: posts.server
    timeout: 800ms

所以，上面的配置，就简化了很多了，这里就不去讲注册中心的实现了，这个不在一个「友好」的 Gateway 的范围之类，在 Gateway 这里，只要能够根据服务的名字发现服务的地址就行了。

热更新

上面只是描述了配置文件的格式，但是配置文件放哪里呢？
配置文件的储存必须要满足如下要求

• 支持热更新
• 能够回滚到特定的版本
• 一条 route 的配置错误，不能影响到全局的配置

首先，要支持特定的版本回滚，你肯定首先想到了 git，是的，配置文件，完全可以放在 git 作为持久化，我们在这里将一些相同业务的路由作为一个项目，所以一个项目可以对应一个 gitlab 的项目，没错，我们选择 gitlab 来储存配置文件。

.
├── config.yaml
└── routes.yaml

0 directories, 2 files

上面就是一个 Gateway 配置的配置（怎么这么绕口），其中 routes.yaml 用来储存上述的各种路由配置，config.yaml 可以用来存一些全局性的配置，比如全局的 timeout 之类的。
然后可以利用 gitlab 的 webhook，当有 push 事件的时候，可以推送到某个地方通知 Gateway 来 gitlab 拉取配置文件，并且更新它。这样就可以实现运行时的 Gateway 配置热更新，而且还可以利用 git 的版本特性，实现回滚版本。

总结

上述还有很多地方待完善，比如 $ 的各个变量的作用域，在内部路由时 $.request 和 $.rpc 又表示什么呢，这些都值得思考和仔细设计。

上篇文章写了 Docker Registry Auth 的流程，这里研究一下 Docker Registry Storage。
Docker 镜像仓库的支持支持多种储存类型，主要分为两大类

• 本地储存
• 远程储存

本地储存为 filesystem 和 inmemory，然后远程储存就主要是一些云计算厂商的储存业务了。无论那种储存都是基于 StorageDriver 这个接口的实现。

储存结构

.
└── docker
    └── registry
        └── v2
            ├── blobs
            │   └── sha256
            │       ├── 07
            │       │   └── 0766572b4bacfaee9a8eb6bae79e6f6dbcdfac0805c7c6ec8b6c2c0ef097317a
            │       ├── 9a
            │       │   └── 9a597e826a59709a4af34279f496c323d496a79e4c998ee5249a738e391192bb
            │       └── 9c
            │           └── 9ca846b27f6e92f0739af5bba5509357b52be0ce0dd02d216f4dccdacd695a8a
            └── repositories
                ├── alpine
                │   ├── _layers
                │   │   └── sha256
                │   │       ├── 0766572b4bacfaee9a8eb6bae79e6f6dbcdfac0805c7c6ec8b6c2c0ef097317a
                │   │       └── 9ca846b27f6e92f0739af5bba5509357b52be0ce0dd02d216f4dccdacd695a8a
                │   ├── _manifests
                │   │   ├── revisions
                │   │   │   └── sha256
                │   │   │       └── 9a597e826a59709a4af34279f496c323d496a79e4c998ee5249a738e391192bb
                │   │   └── tags
                │   │       └── 3.4
                │   │           ├── current
                │   │           └── index
                │   │               └── sha256
                │   │                   └── 9a597e826a59709a4af34279f496c323d496a79e4c998ee5249a738e391192bb
                │   └── _uploads
                └── library
                    └── alpine
                        ├── _layers
                        │   └── sha256
                        │       ├── 0766572b4bacfaee9a8eb6bae79e6f6dbcdfac0805c7c6ec8b6c2c0ef097317a
                        │       └── 9ca846b27f6e92f0739af5bba5509357b52be0ce0dd02d216f4dccdacd695a8a
                        ├── _manifests
                        │   ├── revisions
                        │   │   └── sha256
                        │   │       └── 9a597e826a59709a4af34279f496c323d496a79e4c998ee5249a738e391192bb
                        │   └── tags
                        │       └── 3.4
                        │           ├── current
                        │           └── index
                        │               └── sha256
                        │                   └── 9a597e826a59709a4af34279f496c323d496a79e4c998ee5249a738e391192bb
                        └── _uploads

这个也就是就是官方 registry 镜像的默认储存目录 /var/lib/registry 的文件结构。上述就是镜像 alpine:3.4 在镜像仓库中的储存结构，docker/registry/v2/repositories/library/alpine 表示镜像的 repository，也是我们俗称的镜像，然后在 docker/registry/v2/repositories/library/alpine/_manifests/tags/ 下面是镜像的所有 tag 信息，tag 目录下存的是 tag 的索引，根据索引，再去 docker/registry/v2/blob 目录下就可以一级一级往下找到对应的镜像层了。一般情况下，我们会将镜像分组在相同的 namespace 下，也就是上述的 library，所以其实在官方的 hub 上面，官方的镜像都是数据 library 这个 namespace 下面的，而每个用户就在自己的 namespace 下，也就是 project。

StorageDriver

type StorageDriver interface {
	// Name returns the human-readable "name" of the driver, useful in error
	// messages and logging. By convention, this will just be the registration
	// name, but drivers may provide other information here.
	Name() string

	// GetContent retrieves the content stored at "path" as a []byte.
	// This should primarily be used for small objects.
	GetContent(ctx context.Context, path string) ([]byte, error)

	// PutContent stores the []byte content at a location designated by "path".
	// This should primarily be used for small objects.
	PutContent(ctx context.Context, path string, content []byte) error

	// Reader retrieves an io.ReadCloser for the content stored at "path"
	// with a given byte offset.
	// May be used to resume reading a stream by providing a nonzero offset.
	Reader(ctx context.Context, path string, offset int64) (io.ReadCloser, error)

	// Writer returns a FileWriter which will store the content written to it
	// at the location designated by "path" after the call to Commit.
	Writer(ctx context.Context, path string, append bool) (FileWriter, error)

	// Stat retrieves the FileInfo for the given path, including the current
	// size in bytes and the creation time.
	Stat(ctx context.Context, path string) (FileInfo, error)

	// List returns a list of the objects that are direct descendants of the
	//given path.
	List(ctx context.Context, path string) ([]string, error)

	// Move moves an object stored at sourcePath to destPath, removing the
	// original object.
	// Note: This may be no more efficient than a copy followed by a delete for
	// many implementations.
	Move(ctx context.Context, sourcePath string, destPath string) error

	// Delete recursively deletes all objects stored at "path" and its subpaths.
	Delete(ctx context.Context, path string) error

	// URLFor returns a URL which may be used to retrieve the content stored at
	// the given path, possibly using the given options.
	// May return an ErrUnsupportedMethod in certain StorageDriver
	// implementations.
	URLFor(ctx context.Context, path string, options map[string]interface{}) (string, error)
}

如果要自己实现镜像的 storage，就得按照 StorageDriver 。

对于镜像的 push 过程，最主要的函数就是 StorageDriver.Write，这个主要是用来上传大文件的，而 StorageDriver.PutContent 就是一些小文件的上传，比如说 tag 的 index。

StorageDriver.Write 返回一个 FileWriter，这个也需要自己实现，当有文件上传时，会不停的调用 FileWriter.Write ，当文件上传完毕，会调用 FileWriter.Commit。

Qiniu Storage

官方并没有提供七牛的 storage，但是因为有 StorageDriver，所以也不妨碍我们自己撸一个使用七牛作为储存的私有镜像仓库来

初始化的 driver 的过程，参照阿里云的 OSS 过程，目测阿里云的 OSS 也是参照 AWS 来的，而且阿里云的 OSS API 接口，跟 AWS 真是神似了。

StorageDriver.GetContent

func (d *driver) GetContent(ctx dockerContext.Context, path string) ([]byte, error) {
	baseURL := kodo.MakeBaseUrl(d.domain, d.qiniuPath(path))
	policy := kodo.GetPolicy{}
	privateURL := d.client.MakePrivateUrl(baseURL, &policy)

	resp, err := http.Get(privateURL)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	if resp.StatusCode != 200 {
		return nil, storagedriver.PathNotFoundError{Path: path}
	}

	return ioutil.ReadAll(resp.Body)
}

StorageDriver.GetContent 直接使用七牛的 Golang SDK ，根据 path 生成对应的文件的下载路径，然后下载后，直接返回 []byte 就行了。

StorageDriver.Reader

func (d *driver) Reader(ctx dockerContext.Context, path string, offset int64) (io.ReadCloser, error) {

	contents, err := d.GetContent(ctx, path)
	if err != nil {
		return nil, err
	}
	return ioutil.NopCloser(bytes.NewReader(contents[offset:])), nil
}

StorageDriver.Reader 应该是返回一段 offset 偏移量的文件内容，但是七牛貌似并没有提供像阿里云一样的 range 下载的方式，于是只能将整个文件下载下来，然后再手动切分文件了，这样做，要是镜像的某一层很大，会占用很多内存。

StorageDriver.PutContent

func (d *driver) PutContent(ctx dockerContext.Context, path string, contents []byte) error {
	putExtra := kodo.PutExtra{}
	body := bytes.NewReader(contents)
	err := d.bucket.Put(context.Background(), nil, d.qiniuPath(path), body, int64(len(contents)), &putExtra)
	return err
}

StorageDriver.PutContent 也是直接使用七牛的 SDK 上传小文件。

StorageDriver.Writer

func (d *driver) Writer(ctx dockerContext.Context, path string, append bool) (storagedriver.FileWriter, error) {
	var w *writer
	qiniuPath := d.qiniuPath(path)
	if !append {
		w = &writer{
			key:           qiniuPath,
			driver:        d,
			buffer:        make([]byte, 0, maxChunkSize),
			uploadCtxList: make([]string, 0, 1),
		}
		d.writerPath[qiniuPath] = w
	} else {
		w = d.writerPath[qiniuPath].(*writer)
		w.closed = false

	}
	return w, nil
}

由于七牛的切片上传方式与阿里云提供的切片有点不同，七牛的需要客户端自己记录上传的片段，所以不得不在 StorageDriver 这里做手脚了，这里将 FileWriter 保存在 StorageDriver 上，以便下一次使用。

StorageDriver.Move

func (d *driver) Move(ctx dockerContext.Context, sourcePath string, destPath string) error {
	if err := d.bucket.Move(context.Background(), d.qiniuPath(sourcePath), d.qiniuPath(destPath)); err != nil {
		return err
	}
	return nil
}

StorageDriver.Move 在这里的作用是因为 registry 在上传时，会先将文件放在 docker/registry/v2/repositories/library/alpine/_uploads 目录下面，然后等文件正式上传完毕后，在移动到对应的镜像目录下面。所以也需要 StorageDriver.Delete 删除上传的临时文件。

StorageDriver.Stat

func (d *driver) Stat(ctx dockerContext.Context, path string) (storagedriver.FileInfo, error) {
	entries, commonPrefixes, _, err := d.bucket.List(context.Background(), d.qiniuPath(path), "/", "", 1)
	if err != nil && len(entries) == 0 {

		if err == io.EOF {
			return nil, storagedriver.PathNotFoundError{Path: path}
		}
		return nil, err
	}

	fi := storagedriver.FileInfoFields{
		Path: path,
	}
	if len(entries) == 1 {
		if entries[0].Key != d.qiniuPath(path) {
			fi.IsDir = true
		} else {
			fi.IsDir = false
			fi.Size = entries[0].Fsize
			fi.ModTime = time.Unix(0, entries[0].PutTime)
		}
	} else if len(commonPrefixes) == 1 {
		fi.IsDir = true
	} else {
		return nil, storagedriver.PathNotFoundError{Path: path}
	}
	return storagedriver.FileInfoInternal{FileInfoFields: fi}, nil
}

StorageDriver.Stat 可以在上传时判断文件是否已经存在。

FileWriter.Writer

func (w *writer) Write(p []byte) (int, error) {
	if w.closed {
		return 0, fmt.Errorf("already closed")
	} else if w.committed {
		return 0, fmt.Errorf("already committed")
	} else if w.cancelled {
		return 0, fmt.Errorf("already cancelled")
	}

	if err := w.flushBuffer(); err != nil {
		return 0, err
	}

	w.buffer = append(w.buffer, p...)
	if len(w.buffer) >= maxChunkSize {
		if err := w.flushBuffer(); err != nil {
			return 0, err
		}
	}

	w.size += int64(len(p))

	return len(p), nil
}

func (w *writer) flushBuffer() error {

	for len(w.buffer) >= maxChunkSize {
		if err := w.mkblk(bytes.NewReader(w.buffer[:maxChunkSize]), maxChunkSize); err != nil {
			return err
		}
		w.buffer = w.buffer[maxChunkSize:]
	}
	return nil
}

由于七牛切片上传每块大小最大只能为 4MB ，而 registry 每次上传的一段的大小又是不确定的，所以此处将 registry 每段上传的内容存到 buffer 中，然后当 buffer 里的内容达到 4MB，就将 buffer 中的 4MB 上传到七牛。

FileWriter.Commit

func (w *writer) Commit() error {
	defer func() {
		w.driver.RemoveWriter(w.key)
	}()
	if w.closed {
		return fmt.Errorf("already closed")
	} else if w.committed {
		return fmt.Errorf("already committed")
	} else if w.cancelled {
		return fmt.Errorf("already cancelled")
	}

	if err := w.flushBuffer(); err != nil {
		return err
	}

	if len(w.buffer) > 0 {
		if err := w.mkblk(bytes.NewReader(w.buffer), len(w.buffer)); err != nil {
			return err
		}
	}

	if err := w.mkfile(); err != nil {
		return err
	}

	w.committed = true
	return nil
}

最终上传完毕后，在 FileWriter.Commit 需要将driver 上的FileWriter 给删掉，不然会内存泄露了，在最后 commit 时，再次确认 buffer 中的内容已经全部上传到了七牛，然后调用七牛的 mkfile 合并块成文件，想当初去七牛面试技术支持的岗位时，还问过我这个问题，当时答得并不咋地。

最后

最终需要在 cmd/registry/main.go 的文件中初始化我们自己的 driver。

完整的七牛 Storage 在这里，https://gist.github.com/cloverstd/813a0df6b9eacc459dd84509e902e744 。

原理简介

Bamboo 为了 DO/CS 系统提供了服务自动发现的功能，当 Marathon 上部署了一个应用，Bamboo 能够将应用下所有的容器的端口更新到 HAProxy 的配置文件中，并且 reload HAProxy，从而实现应用的服务自动发现，并且利用 HAProxy 做应用中容器的负载均衡。

Bamboo 主要的原理就是利用 Marathon 的 Event-Bus API，当 Marathon 上的应用发生改变（scaling）时，Event-Bus 会通知对应的 Subscriber (Bamboo)，然后 Subscriber 会做出对应的操作。

除了 Bamboo，还有 Marathon-lb 也是类似的原理。

源码分析

Bamboo 是使用 Golang 开发的，代码量很少，原理也不复杂。

首先是从配置文件读取配置，然后读取对应的环境变量，当环境变量存在时，覆盖相应的配
置项，对应的代码在 configuration#L44 。

然后就是声明一个 EventBus，用来将 Marathon、ZK 发生改变时的动作传递到对应的 Handler 函数中去操作。

在 bamboo.go#L80-L81 这里注册了两个主要的 EventHandler，MarathonEventHandler 对应的是 Marathon 上的应用发生改变时的动作，ServiceEventHandler 对应的是 ZK 上的应用配置发生改变时的动作。

在这里，会使用 listenToZookeeper 监听 ZK 的改变，然后应用到ServiceEventHandler上。

最后启动一个 HTTPServer 作为对 ZK 操作的 API 接口，对于 Marathon 事件的监听，Bamboo 支持 Marathon 的两种方式

• 主动调用
• 回调模式

主动调用的过程是首先发起一个 HTTP 请求到 Marathon 的 /v2/events 接口，这个接口是一个 stream 的 HTTP 请求，Marathon 会将应用的改变 push 到建立的连接中。回调模式是 Bamboo 向 Marathon 注册自己的地址，当 Marathon 上的应用发生改变时，会向 Bamboo 注册的地址发送 HTTP 请求，通知 Bamboo。

这两种模式可以通过环境变量 MARATHON_USE_EVENT_STREAM 来启用主动调用的模式，主动调用在本地开发调试 Bamboo 时很有用，这样就不需要 Marathon 回调本机地址了。

无论是回调模式还是主动调用，当 Marathon 的发生改变时，最终都会调用 EventBus.Publish 的方法，而 EventBus.Publish 就会遍历注册的 Handler，然后调用对应的事件的 Handler。

主要两个 Handler 就是上述的 MarathonEventHandler 和 ServiceEventHandler。

在应用启动时会调用 event_handler 的 init 函数，在 init 函数中，通过 updateChan 这个带有 buffer 的 channel 与 handler 进行通信，并且同时也创建了一个带 buffer 的 channel queueUpdateSem，容量为 1，用来保证同一时间只有一个 HAProxy 在更新，并且后续的一个更新操作可以排队，再之后的更新操作就被 block 掉了。

上述的两个 Handler 都会触发更新 HAProxy 的动作，而更新 HAProxy 的流程是，首先通过 ensureLatestConfig 函数确认 HAProxy 配置文件是否有更新的必要，原理就是生成新的 HAProxy 配置，然后读取旧的 HAProxy，对比配置文件是否一致，如果一模一样就放弃此次更新。当确认要更新之后，会通过ReloadValidationCommand定义的命令来检验 HAProxy 的配置是否正确，只有校验通过的配置，才能调用ReloadCommand通知 HAProxy 重新加载新的配置文件。

Bamboo 通过 Golang 的 text/template 来生成 HAProxy 配置文件，语法就是 Golang 的模板语法，然后 Bamboo 加了一些便捷函数。

一些问题

• Marathon 上运行多个应用时，并且都通过 Bamboo 配置了 HAProxy 的 ACL，如果一个应用更新时，其他应用的访问会在很短的时间内中断，对于这个问题，可以通过类似于 Marathon-lb 的 HAPROXY_GROUP 的方法，将 Bamboo 绑定到单独的应用组上。
• 根据客户反馈，HAProxy 在 reload 的过程中，会发生一定几率的中断，yelp 有一篇文章分析这个问题，可以实现 zero-downtime 的 HAProxy reload，然后数人云写了这个据说这可以『实现无中断 Reload Haproxy』。

最近遇到了 Docker 镜像体积过大的问题，对于部署、移交相当麻烦和慢。于是就抽点时间研究下了怎么减小 Docker 镜像的体积。

下面我以手动编译 nginx 镜像作为例子来减小镜像体积，nginx 的版本是 1.10.2，并且下载到了当前目录。

下图是各种 Dockerfile 制作出镜像的效果

• 替换基础镜像
  • 基于 centos 的镜像
  • 基于 Alpine 的镜像
• 移除 build 依赖的文件
  • centos 基础镜像的依赖
  • Alpine 基础镜像的依赖
• 减少镜像层级
  • centos
  • Alpine
  • 大文件测试
    • 通过 COPY
    • 通过 wget
• 其他
  • Python
  • 通过镜像层级加速 build 过程

FROM centos:7.2.1511
COPY . /tmp/

RUN yum install -y make \
    gcc \
    openssl-devel \
    zlib-devel \
    perl-devel \
    pcre-devel && \
    cd /tmp/ && tar zxf nginx-1.10.2.tar.gz && \
    cd /tmp/nginx-1.10.2 && \
    ./configure --with-http_gzip_static_module --with-http_ssl_module && \
    make && make install && \
    rm -rf /tmp/* && \
    ln -sf /dev/stdout /usr/local/nginx/logs/access.log && \
    ln -sf /dev/stderr /usr/local/nginx/logs/error.log

EXPOSE 80
CMD ["/usr/local/nginx/sbin/nginx", "-g", "daemon off;"]

上述就是一个最基本的 nginx 镜像了，通过 docker build 后，镜像大小是 535.2MB。nginx 的源代码才 890KB，这种镜像，完全是浪费空间。

FROM alpine:3.4
COPY . /tmp/

RUN apk update && \
    apk add gcc make openssl-dev zlib-dev perl-dev pcre-dev libc-dev && \
    cd /tmp/ && tar zxf nginx-1.10.2.tar.gz && \
    cd /tmp/nginx-1.10.2 && \
    ./configure --with-http_gzip_static_module --with-http_ssl_module && \
    make && make install && \
    rm -rf /tmp/* && \
    ln -sf /dev/stdout /usr/local/nginx/logs/access.log && \
    ln -sf /dev/stderr /usr/local/nginx/logs/error.log

EXPOSE 80
CMD ["/usr/local/nginx/sbin/nginx", "-g", "daemon off;"]

通过上述的 Dockerfile docker build 出来的镜像大小是 155.8MB，体积已经缩小到了 3 倍以上了，对于 nginx 的运行来说，centos 和 Alpine 完全是一样的效果，所以对于基础镜像，首选应该是 Alpine 这之类的精简版系统，而且在官方 hub 上的官方镜像，大多有提供 Alpine 版的版本。

COPY . /tmp/

RUN yum install -y make
gcc
openssl-devel
zlib-devel
perl-devel
pcre-devel &&
cd /tmp/ && tar zxf nginx-1.10.2.tar.gz &&
cd /tmp/nginx-1.10.2 &&
./configure --with-http_gzip_static_module --with-http_ssl_module &&
make && make install &&
yum remove -y gcc openssl-devel zlib-devel perl-devel pcre-devel make &&
yum clean all &&
rm -rf /tmp/* &&
ln -sf /dev/stdout /usr/local/nginx/logs/access.log &&
ln -sf /dev/stderr /usr/local/nginx/logs/error.log

EXPOSE 80
CMD ["/usr/local/nginx/sbin/nginx", "-g", "daemon off;"]

上面的 `yum remove -y gcc openssl-devel zlib-devel perl-devel pcre-devel make` 和 `yum clean all` 就是移除不需要的文件，从而进一步减小体积，加上两条命令后，执行 `docker build` 后的镜像的体积为 413.8MB，相对上述的 535.2MB 减小了 121.4 MB。

<div id="build-dependence-file-alpine"></div>
## Alpine 基础镜像的依赖
上述 centos 的移除依赖的后体积变化可能看着没有啥感觉，因为本身就很大，下面是 Alpine 的 Dockerfile

```Dockerfile
FROM alpine:3.4

COPY . /tmp/

RUN apk update && \
    apk add gcc make openssl-dev zlib-dev perl-dev pcre-dev libc-dev && \
    cd /tmp/ && tar zxf nginx-1.10.2.tar.gz && \
    cd /tmp/nginx-1.10.2 && \
    ./configure --with-http_gzip_static_module --with-http_ssl_module && \
    make && make install && \
    apk del gcc make openssl-dev zlib-dev perl-dev pcre-dev libc-dev && \
    rm -rf /var/cache/* && \
    rm -rf /tmp/* && \
    ln -sf /dev/stdout /usr/local/nginx/logs/access.log && \
    ln -sf /dev/stderr /usr/local/nginx/logs/error.log

EXPOSE 80
CMD ["/usr/local/nginx/sbin/nginx", "-g", "daemon off;"]

这次 docker build 出来的镜像的体积为 11.74MB，直接从 155.8MB 减小了 144.04 MB，达到了数量级的变化。并且这个 11.74MB 的镜像也是可以正常运行的。

那么对于一些需要通过 COPY 命令的方式拷贝到镜像里面的文件，可以使用 wget 的方式，下载到镜像里，然后使用完之后就删掉。

RUN yum install -y make
gcc
openssl-devel
zlib-devel
perl-devel
pcre-devel
wget &&
wget --no-check-certificate -O /tmp/nginx-1.10.2.tar.gz http://nginx.org/download/nginx-1.10.2.tar.gz &&
cd /tmp/ && tar zxf nginx-1.10.2.tar.gz &&
cd /tmp/nginx-1.10.2 &&
./configure --with-http_gzip_static_module --with-http_ssl_module &&
make && make install &&
rm -rf /tmp/* &&
ln -sf /dev/stdout /usr/local/nginx/logs/access.log &&
ln -sf /dev/stderr /usr/local/nginx/logs/error.log

EXPOSE 80
CMD ["/usr/local/nginx/sbin/nginx", "-g", "daemon off;"]

上述通过 wget 的下载方式将 nginx 的源代码下载到了镜像里面，然后使用完之后，再删除掉，这样对于镜像来说，将不会存在 nginx 的源代码。通过此方法 `docker build` 出来的镜像是 534.8MB，相对于最原始的 535.2MB，少了一丢丢，这种方式随着文件的大小增大，效果显著。

<div id="level-alpine"></div>
## Alpine
```Dockerfile
FROM alpine:3.4

RUN apk update && \
    apk add gcc make openssl-dev zlib-dev perl-dev pcre-dev libc-dev wget && \
    wget --no-check-certificate -O /tmp/nginx-1.10.2.tar.gz http://nginx.org/download/nginx-1.10.2.tar.gz && \
    cd /tmp/ && tar zxf nginx-1.10.2.tar.gz && \
    cd /tmp/nginx-1.10.2 && \
    ./configure --with-http_gzip_static_module --with-http_ssl_module && \
    make && make install && \
    rm -rf /tmp/* && \
    ln -sf /dev/stdout /usr/local/nginx/logs/access.log && \
    ln -sf /dev/stderr /usr/local/nginx/logs/error.log

EXPOSE 80
CMD ["/usr/local/nginx/sbin/nginx", "-g", "daemon off;"]

通过 docker build 后，镜像大小为 155.4MB，因为 Alpine 的体积本身很小，所以效果不明显。

COPY . /code
RUN rm -rf /code

CMD ["tail", "-f", "/etc/hosts"]

基础镜像采用的是 Alpine，对于 2.3GB 来说，基础的镜像的体积可以忽略。`docker build` 出来的镜像体积是 2.463GB，可以看到，虽然通过 `rm -rf /code` 将文件删除了，但是由于 docker 镜像是分层的原因，所以镜像的依旧包含了 COPY 进去的文件的体积，虽然文件在 `docker run` 时是看不到的。

<div id="bigfile-by-wget"></div>
### 通过 wget
```Dockerfile
FROM alpine:3.4

RUN apk update && apk add wget && \
    mkdir /code && \
    wget --no-check-certificate -O /code/test.dbf http://172.24.4.188:8000/test.dbf && \
    apk del wget && \
    rm -rf /var/cache/* && \
    rm -rf /code

CMD ["tail", "-f", "/etc/hosts"]

通过此方法 docker build 出来的镜像体积为 4.817MB，相对于 Alpine 的基础镜像 4.799MB，只增加了 0.018MB。相对于上述的 COPY 完全是质的变化。

通过 find / -name "*.py[co]" -exec rm '{}' ';' 就可以全都移除掉

从开始会写程序开始，一直都很讨厌处理上传文件，因为总觉得在自己写的 controller 里面从 HTTP Body 里读文件，然后再写到本地磁盘的过程是一件非常不（易）靠（出）谱（错）的事，总怕文件没有正确写入，而且常用的框架对于上传文件，都是直接将整个文件加载到内存中，当处理大文件上传时，内存岂不是要爆炸了。

后来知道了七牛和又拍云之类的云服务，顿时感觉对于文件上传的过程的瞬间简化了。类似与七牛之类的文件储存云服务一般都会提供文件直传的接口，直接在前端就把文件上传到了七牛的服务器，而不经过业务服务器，然后当文件上传成功后，可以选择让七牛通知（回调）到业务服务器。

但是一般只有『互联网』企业才会选择七牛之类的云服务，可惜我就没怎么待过『互联网』企业，所以一直都没有正式的使用过文件直传这种方式，甚是伤心。如果你没有用过七牛，强烈推荐你试用一下，这里有我的邀请链接，你注册成功，我将收到 40G 的免费流量 https://portal.qiniu.com/signup?code=3lpwnduxidob5 ，然后这里是没有邀请码的链接 https://portal.qiniu.com/signup 。

然后最近做的项目有部分需要文件上传，于是就研究了下 Nginx Upload Module 的使用方法，目测七牛也是用的类似的方法。

• Nginx Upload Module
  • 安装
  • 配置
  • 上传参数
• 业务程序处理文件
  • 使用 Tornado 来处理文件
• 参考

Nginx Upload Module

Nginx Upload Module 是 Nginx 的一个模块，一般 Linux 发行版里的 Nginx 包都没有预装这个模块，所以需要自己手动编译 Nginx，并且加入这个模块。

安装

手动编译 Nginx 需要一些依赖包，本着极简主义的原则，只安装必须的依赖包

• openssl Nginx Upload Module 会用到
• pcre Nginx Rewrite 会用到
• zlib gzip 压缩会用到

wget https://www.openssl.org/source/openssl-1.0.2j.tar.gz
wget ftp://ftp.csx.cam.ac.uk/pub/software/programming/pcre/pcre-8.39.tar.gz
wget http://zlib.net/zlib-1.2.8.tar.gz

然后是下载最重要的 Nginx 了，这里我选择的 Nginx 是 1.10.2，目前的最新稳定版，然后 Nginx Upload Module，看 git 记录，已经至少有 6 年没更新过了，从 Nginx Upload Module 的 Github 直接下载的最新版，在新版本（1.8.x）以上没法编译成功，需要用2.2这个分之的代码

wget http://nginx.org/download/nginx-1.10.2.tar.gz
git clone -b 2.2 https://github.com/vkholodkov/nginx-upload-module

然后就是安装了

tar zxf openssl-1.0.2j.tar.gz
tar zxf pcre-8.39.tar.gz
tar zxf zlib-1.2.8.tar.gz
tar zxf nginx-1.10.2.tar.gz
cd nginx-1.10.2

./configure --add-module=../nginx-upload-module --with-openssl=../openssl-OpenSSL_1_0_2j --with-http_gzip_static_module --with-http_ssl_module --with-pcre=../pcre-8.39/ --with-zlib=../zlib-1.2.8
make && sudo make install

如果一切顺利的话，Nginx 已经被默认安装到了 /usr/local/nginx/ 目录下了。

配置

上述完成后，通过 sudo /usr/local/nginx/sbin/nginx 就可以启动 Nginx 了。接下来是配置 Nginx Upload Module 了，具体的配置项，在官方文档 已经说得很清楚了

server {
    client_max_body_size 100m;
    listen       80;

    # Upload form should be submitted to this location
    location /upload {
        # Pass altered request body to this location
        upload_pass   @test;

        # Store files to this directory
        # The directory is hashed, subdirectories 0 1 2 3 4 5 6 7 8 9 should exist
        upload_store /tmp 1;

        # Allow uploaded files to be read only by user
        upload_store_access user:r;

        # Set specified fields in request body
        upload_set_form_field $upload_field_name.name "$upload_file_name";
        upload_set_form_field $upload_field_name.content_type "$upload_content_type";
        upload_set_form_field $upload_field_name.path "$upload_tmp_path";

        # Inform backend about hash and size of a file
        upload_aggregate_form_field "$upload_field_name.md5" "$upload_file_md5";
        upload_aggregate_form_field "$upload_field_name.size" "$upload_file_size";

        upload_pass_form_field "^submit$|^description$";

        upload_cleanup 400 404 499 500-505;
    }

    # Pass altered request body to a backend
    location @test {
        proxy_pass   http://localhost:8080;
    }
}

Nginx 安装好后，默认的用户为 nobody，如果不修改，Nginx Upload Module 上传的文件所属是 nobody，其他用户没权限动它，要么是将 Nginx 的 user 配置修改，要么修改 Nginx Upload Module 的 upload_store_access 为 all:rw，不然没法读写。
接下来是在 upload_store 设置的目录下创建对应的是个文件夹，从 0 到 9，注意确保 Nginx 有权限操作这 10 个目录

mkdir -p /tmp/0 /tmp/1 /tmp/2 /tmp/3 /tmp/4 /tmp/5 /tmp/6 /tmp/7 /tmp/8 /tmp/9

其中 upload_store 后面跟随的数字表示 Nginx Upload Module 存的文件的目录的级数，如果为 2 则表示上述的 10 个目录下再来 10 个目录，同样要自己创建。

然后重新加载 Nginx 的配置文件

sudo /usr/local/nginx/sbin/nginx -s reload

如果没意外 Nginx Upload Module 就正式安装好了。

上传参数

前端的上传参数如下

<form name="upload" method="POST" enctype="multipart/form-data" action="/upload">
      <input type="file" name="file"><br>
      <input type="file" name="file"><br>
      <input type="submit" name="submit" value="Upload">
    </form>

业务程序处理文件

Nginx Upload Module 只做一件事，那就是上传文件存到对应的地方，文件直接存起来是没啥用的，还需要跟真正的业务系统结合起来，所以 Nginx Upload Module 提供了一个 directive upload_pass 来将上传成功后的信息发送给业务程序，并且可以通过 upload_set_form_field 和 upload_aggregate_form_field 将前端的一些参数一起传递给后端去，这样真是太方便了。

使用 Tornado 来处理文件

class UploadHandler(tornado.web.RequestHandler):

    @tornado.gen.coroutine
    def post(self):
        keys = self.request.arguments.keys()
        if "file.path" not in keys:
            self.set_status(status_code=400, reason="file field not exist.")
            self.write("400")
            return
        if filter(lambda x: not x.startswith("file."), keys):
            self.set_status(status_code=400, reason="only allow file field upload")
            self.write("400")
            return
        files = list()
        file_path = self.request.arguments['file.path']
        for index in xrange(len(file_path)):
            file = {}
            file['name'] = self.request.arguments['file.name'][index]
            file['content_type'] = self.request.arguments['file.content_type'][index]
            file['path'] = self.request.arguments['file.path'][index]
            file['md5'] = self.request.arguments['file.md5'][index]
            file['size'] = self.request.arguments['file.size'][index]
            files.append(file)
        # mv tmp file to save store storage
        for file in files:
            # debug
            src_file = file['path']
            mime = magic.from_file(src_file, mime=True)
            ext = mimetypes.guess_extension(mime, False)
            dest_file = os.path.join(
                options.storage_path,
                file['md5'] + ext
            )
            if not os.path.exists(dest_file):
                yield LinuxUtils.mv(
                    src_file,
                    dest_file
                )
            else:
                yield LinuxUtils.rm(
                    src_file
                )
            file['path'] = file['md5'] + ext
        self.write({"data": files})

上面是处理文件的示例，主要是做两件事

• 配合 Nginx Upload Module 的 upload_cleanup directive 过滤掉不正确的文件上传参数
• 将文件从临时目录移到其他目录，并且使用 libmagic 来判断文件的类型，从而重命名文件

完整的代码在这里 https://gist.github.com/cloverstd/deef6e9a4db76dab5c4d33ee68f63ec5

从此，虽然用不了七牛，但是文件上传也可以成为一件愉快的事了。

参考

• https://www.nginx.com/resources/wiki/modules/upload/
• https://imququ.com/post/my-nginx-conf.html
• http://agentzh.org/misc/code/nginx/core/ngx_file.c.html#L430

由于工作原因，最近研究了下 Docker Registry V2 （以下内容如无特别说明，Registry 均指 Docker Registry V2）的私有化（这里所指的私有化并不是指的私有仓库，而是在私有云里面面向内部的共有仓库，并且对外隔离的）的定制化开发。

Registry 是 Docker 官方提供的镜像仓库，官方的 Docker Hub 就是基于 Registry 的，Registry 分为 V1 和 V2，V1 是我大爱的 Python 的写的，V2 是我现在大爱的 Golang 开发的，由于 V1 已经被官方抛弃了，所以目前的定制化是基于 V2 版本的。

• 认证结构
• 认证流程
  • 开启 Registry 认证模式
    • 生成密钥
    • 启动 Registry
  • Auth Server
    • 生成 jwt token
      • Claim
      • Headers
      • token
    • 响应 Docker 客户端
  • 具体实现代码
• 参考

Registry 提供了松耦合的 Registry 系统，将 Auth Server 与 Registry 的 push/pull 给分离开来了，将 Auth Server 的实现交由用户来实现，个人觉得真是太赞了。

上图是官方认证结构图，用户在启动 Registry 并且配置 auth.token 的认证方式时，当用户进行 push/pull/login 的操作时，会到 Auth Server 去认证，由 Auth Server 来判断是否授权通过。基于这一模式，就可以在私有化 Registry 中控制不同的用户（项目）有不同的权限。

• auth.token.issuer : token 发行人，用来在 Auth Server 识别身份
• auth.token.realm : Auth Server 的 URI
• auth.token.rootcertbundle : 公钥的绝对路径
• auth.token.service : 认证的 service，当多个 Registry 公用一个 Auth Server 时，可以区分来自哪里

上述的 auth.token.rootcertbundle 可以使用 openssl 来生成，在 Token 认证体系里是需要自己生成私钥和公钥来做 token 的认证的。

# 生成私钥
openssl genrsa -out ./private_key.pem 4096
# 生成公钥
openssl req -new -x509 -key ./private_key.pem -out ./root.crt -days 3650 -subj /C=CN/ST=state/L=CN/O=cloverstd/OU=cloverstd\ unit/CN=hui.lu/emailAddress=i@hui.lu

上述操作后，会在当前目录生成 private_key.pem 和 root.crt

在启动后可以试下 docker pull 127.0.0.1:5000/nginx 这个命令，Docker 会返回 getsockopt: connection refused 的错误，因为我们的 Auth Server 还没有启动起来。

当执行 pull/login/push 这三个需要认证的命令时，Auth Server 会收到一条 GET 请求，并且会有一些参数。

在上述配置中，我配置的 auth.token.realm 是 http://11.11.11.1:8080/service/token，当我执行 docker pull 127.0.0.1:5000/nginx 时，Auth Server 会收到下面这样的请求

GET /service/token?scope=repository:nginx:pull&service=token-service HTTP/1.1
Content-Length:
User-Agent: docker/1.12.0 go/go1.6.3 git-commit/8eab29e kernel/4.4.15-moby os/linux arch/amd64 UpstreamClient(Docker-Client/1.12.0 (darwin))
Connection: close
Host: 11.11.11.1:8080
Content-Type:
Accept-Encoding: gzip

这是因为 Docker 首先会发一条请求过来确认是否需要认证，这里服务器要给正确的相应，告诉 Docker 是否需要认证，正确的要求 Docker 提供认证服务的请求应该如下

HTTP/1.1 401 Unauthorized
Content-Type: application/json; charset=utf-8
Docker-Distribution-Api-Version: registry/2.0
Www-Authenticate: Bearer realm="http://11.11.11.1:8080/service/token",service="token-service",scope="repository:nginx:pull,push"
Date: Thu, 10 Sep 2015 19:32:31 GMT
Content-Length: 235
Strict-Transport-Security: max-age=31536000

{"errors":[{"code":"UNAUTHORIZED","message":"access to the requested resource is not authorized","detail":[{"Type":"repository","Name":"samalba/my-app","Action":"pull"},{"Type":"repository","Name":"samalba/my-app","Action":"push"}]}]}

这里有以下俩要注意

• Www-Authenticate 这里要指出 Auth Server 的地址，还有支持的 scope 操作
• 在 body 里需要给出错误的原因
• HTTP Status Code 必须为 401

其实我测了下发现，并没那么严格的要求，只要 HTTP Status Code 为 401 并且 body 的格式相对正确就够了。

然后，Docker 会提示输入用户名和密码，这里我输入 test 和 test，然后服务器会收到跟上述类似的请求，只是多了一个 HTTP Basic Auth 的头 Authorization。通过 base64 解码一下就可以拿到用户名和密码，然后在这里就可以通过各种验证手段（数据库查询等）来验证用户的身份信息了，身份信息认证过了之后，就要认证用户是否有权操作他所要操作的 scope 了

scope 根据官方的说明，可能会有多个 scope 作为 URL 参数传送过来，所以这里需要获取到的是一个 scope 的数组，在很多框架里都有实现获取同名参数，作为一个数组。

scope 由 Name、Type 和 Actions 组成，并且由 : 分割开来，直接 split(':') 一下就可以得到解析过后的 scope 了，一般来说，scope 会是以下的样子

{
    "scopes": [{
        "type": "repository",
        "name": "nginx",
        "actions": ["pull", "push"]
    }]
}

然后就可以根据 name 和 actions 来选择用户是否有权操作了，在 Registry 里并没有很明确的用户私有空间这样的概念，通常我们所指定 Registry，例如 127.0.0.1:5000/test/nginx:1.9 是指在 127.0.0.1:5000 这个 Registry 里面，test 用户下的 nginx 镜像的 1.9 版本，其实在 Registry 里，并没有 test 这个用户空间，这里的是 test 完全是 Auth Server 可以自定义的一种命名空间而已，在 Resgitry 里，/test/nginx:1.9 就相当于一段唯一的路由而已。

• iss 前期通过 registry.yaml 配置文件里的 issuser
• sub 当前操作的帐号或者说用户名
• aud 前期通过 registry.yaml 配置文件里的 service
• exp、iat 和 nbf 表示的是 token 时间相关的
• jti 是一段随机字符串
• access 就是上述解析过后的 scope

der_public_key = public_key.public_bytes(Encoding.DER, PublicFormat.SubjectPublicKeyInfo)

sha256 = hashlib.sha256(der_public_key)
base32_payload = base64.b32encode(sha256.digest()[:30]) # 240bits / 8
res = ":".join(
[base32_payload[i:i+4] for i in xrange(0, 48, 4)]
)

最后的 `res` 也就是 `kid` 应该是这样的类型的 `PYYO:TEWU:V7JH:26JV:AQTZ:LJC3:SXVJ:XGHA:34F2:2LAQ:ZRMK:Z7Q6`。

<div id="jwt-token"></div>
### token
拿到 `kid` 和 `claim` 后就可以来生成 token 了
```python
token = jwt.encode(claim, private_key, algorithm='RS256', headers=headers)

app = Flask(name)

SERVICE = 'token-service'
ISSUER = 'registry-token-issuer'

class DockerRegistryAuth(object):

def __init__(self, account, scopes, service, issuer, private_key_path, token_expires=300):
    self.account = account
    self.issuer = issuer
    self.scopes = scopes
    self.access = self.get_access_by_scopes(scopes)
    self.service = service
    self.private_key_path = private_key_path
    self.token_expires = token_expires

@property
def private_key(self):
    if getattr(self, '_private_key_content', None):
        return self._private_key_content

    with open(self.private_key_path, 'r') as fp:
        setattr(self, '_private_key_content', fp.read())
        return self._private_key_content

@property
def public_key(self):
    private_key = load_pem_private_key(
            self.private_key,
            password=None,
            backend=default_backend()
    )
    _public_key = private_key.public_key()
    return _public_key

def check_service(self, service):
    return self.service == service

def get_token(self):
    now = int(time.time())

    claim = {
        'iss': self.issuer,
        'sub': self.account,
        'aud': self.service,
        'exp': now + self.token_expires,
        'nbf': now,
        'iat': now,
        'jti': base64.b64encode(os.urandom(1024)),
        'access': self.access
    }

    headers = {
        'kid': self.get_kid()
    }
    token = jwt.encode(claim, self.private_key, algorithm='RS256', headers=headers)
    return {
        'token': token,
        'issued_at': now,
        'expires_in': now + self.token_expires
    }

def get_kid(self):
    der_public_key = self.public_key.public_bytes(Encoding.DER, PublicFormat.SubjectPublicKeyInfo)

    sha256 = hashlib.sha256(der_public_key)
    base32_payload = base64.b32encode(sha256.digest()[:30]) # 240bits / 8
    return ":".join(
        [base32_payload[i:i+4] for i in xrange(0, 48, 4)]
    )


def get_login_info(self, authorization):
    if not authorization:
        return None
    auth_info = authorization
    if authorization.startswith('Basic'):
        auth_info = authorization[5:]

    user_info = base64.b64decode(auth_info)
    self.username, self.password = user_info.split(':')
    return {
        'username': self.username,
        'password': self.password,
    }

def get_access_by_scopes(self, scopes):
    access = list()
    if not scopes:
        return access
    for scope in scopes:
        type_, name, actions = scope.split(':')
        access.append({
            'type': type_,
            'name': name,
            'actions': actions.split(',')
        })

    return access

def unauthorized401(access, message=None, code=None):
detail = list()
for scope in access:
for action in scope['actions']:
detail.append({
"Action": action,
"Name": scope['name'],
"Type": scope['type']
})
data = {
"errors": [
{
"code": code or "UNAUTHORIZED",
"detail": detail,
"message": message or "access to the requested resource is not authorized"
}
]
}
resp = make_response(json.dumps(data), 401)
resp.headers['Content-Type'] = 'application/json; charset=utf-8'
resp.headers['Docker-Distribution-Api-Version'] = 'registry/2.0'
resp.headers['Www-Authenticate'] = 'Bearer realm="https://auth.docker.io/token",service="registry.docker.io",scope="repository:samalba/my-app:pull,push"'
return resp

@app.route('/service/token')
def service_token():
print request.headers
scopes = request.args.getlist('scope')
account = request.args.get('account')
client_id = request.args.get('client_id')
service = request.args.get('service')
http_base_auth = request.headers.get('Authorization')

registry_auth = DockerRegistryAuth(account, scopes, SERVICE, ISSUER, './private_key.pem')
if not registry_auth.check_service(service):
    return unauthorized401(registry_auth.access, 'service not be allowed.')
user = registry_auth.get_login_info(http_base_auth)

if user:
    if not (user['username'] == 'test' and user['password'] == 'test'):
        return unauthorized401(registry_auth.access, 'incorrect username or password')
    res = registry_auth.get_token()
    return jsonify(
        token=res['token'],
    )

return unauthorized401(registry_auth.access)

if name == 'main':
app.run(host='0.0.0.0', port=8080, debug=True)

<div id="ref"></div>
# 参考
* https://github.com/vmware/harbor
* https://docs.docker.com/registry
* https://github.com/itrp/docker-token-auth
* https://cryptography.io/en/latest/hazmat/primitives/asymmetric/rsa/
* https://pyjwt.readthedocs.io/en/latest/

因为工作项目的需求，研究了一下 WebSSH。

• 实现原理
• 通过 Docker cli 运行 bash
• 通过 Docker API 运行 bash
  • 创建一个 exec 实例
  • 执行 exec 实例
• 与前端结合实现 WebSSH
• 参考

• POST /containers/(id or name)/exec
• POST /exec/(id)/start

首先是在一个容器里创一个 exec 的实例，然后是执行这个 exec 的实例。

但是这里的 /exec/(id)/start，Docker 用到了一个 tcp stream 来交互 stdin、stdout 和 stderr。其实官方的 Python SDK 里是有这个的实现，但是我用的是 Tornado，并没有找到相关的库，于是乎又只能撸轮子了。

下面是一步一步分析，这里假设容器 ID 为 'XXXX'

这个就是一个简单的 POST 请求。

http_client = tornado.httpclient.AsyncHTTPClient()
res = yield http_client.fetch(
    'http://{host}:{port}/containers/{container}/exec'.format(host=host, port=port, container=container),
    method='POST',
    headers={
        'Content-Type': 'application/json',
    },
    body=json.dumps({
        'AttachStdin': True,
        'AttachStdout': True,
        'AttachStderr': True,
        'DetachKeys': 'ctrl-p,ctrl-q',
        'Tty': True,
        'Cmd': [
            '/bin/bash'
        ]
    })
)
data = res.body
print data
exec_id = json.loads(data)['Id']

通过 /containers/(id or name)/exec 请求，得到一个 exec id。

在 Docker 这个 API 实现这里，跟 WebSocket 的实现类似，首先是一个 POST 请求过来（WebSocket 是 GET 请求），然后服务器不会中断这个连接，而是返回给客户端下面这样的 HTTP 报文

HTTP/1.1 200 OK
Content-Type: application/vnd.docker.raw-stream

如果客户端的请求头里有下面两个头的话（其实只需要一个 Upgrade）

Upgrade: tcp
Connection: Upgrade

Docker API 服务器会返回下面这样的 HTTP 报文

HTTP/1.1 200 OK
Content-Type: application/vnd.docker.raw-stream
Connection: Upgrade
Upgrade: tcp

然后 Docker API 服务器就会往这个 tcp stream 里写 stdout 和 stderr（如果请求的时候有这个的需求）的内容，然后也可以往这个 tcp stream 里发送数据到容器里执行的命令的 stdin（如果请求的时候有这个需求）里面。

下面是这个操作的代码

tcp_client = tornado.tcpclient.TCPClient()
conn = yield tcp_client.connect(host, port)
data = json.dumps({
    'Detach': False,
    'Tty': True
})
yield conn.write(b'POST /exec/{}/start HTTP/1.1\r\n'.format(exec_id))
yield conn.write(b'Host: 11.11.11.3:2375\r\n')
yield conn.write(b'Connection: Upgrade\r\n')
yield conn.write(b'Content-Type: application/json\r\n')
yield conn.write(b'Upgrade: tcp\r\n')
yield conn.write(b'Content-Length: {}\r\n'.format(len(data)))
yield conn.write(b"\r\n")
yield conn.write(data.encode('utf-8'))
res = yield conn.read_until(b'\r\n\r\n')

这里为什么要手动构建 HTTP 请求呢？因为没找到 Tornado 的 HTTPClient 有这种协议的实现，反正我是没找到。

然后就可以通过 conn 来读写数据了，conn 是一个 IOStream 的实例。

具体的实现原理是打开前端页面，与后端建立一个 WebSocket 的连接，然后后端在打开 WebSocket 的连接的同时，通过上述的通过 Docker API 运行 bash 建立一个 tcp 连接，并且把前端所有的输入都转发到 tcp 去，然后 tcp 所有的输出都转发到 WebSocket 去。

具体后端代码如下

class BashHandler(tornado.websocket.WebSocketHandler):

    @tornado.gen.coroutine
    def open(self):
        print "WebSocket opened"
        container = '0e9bfdfe4639'
        container = '4798e46ea3c'
        host = '11.11.11.3'
        port = 2375
        host = '172.24.6.171'
        port = 4000
        http_client = tornado.httpclient.AsyncHTTPClient()
        res = yield http_client.fetch(
            'http://{host}:{port}/containers/{container}/exec'.format(host=host, port=port, container=container),
            method='POST',
            headers={
                'Content-Type': 'application/json',
            },
            body=json.dumps({
                'AttachStdin': True,
                'AttachStdout': True,
                'AttachStderr': True,
                'DetachKeys': 'ctrl-p,ctrl-q',
                'Tty': True,
                'Cmd': [
                    '/bin/bash'
                ]
            })
        )
        data = res.body
        print data
        exec_id = json.loads(data)['Id']
        print exec_id
        tcp_client = tornado.tcpclient.TCPClient()
        conn = yield tcp_client.connect(host, port)
        data = json.dumps({
            'Detach': False,
            'Tty': True
        })
        yield conn.write(b'POST /exec/{}/start HTTP/1.1\r\n'.format(exec_id))
        yield conn.write(b'Host: 11.11.11.3:2375\r\n')
        yield conn.write(b'Connection: Upgrade\r\n')
        yield conn.write(b'Content-Type: application/json\r\n')
        yield conn.write(b'Upgrade: tcp\r\n')
        yield conn.write(b'Content-Length: {}\r\n'.format(len(data)))
        yield conn.write(b"\r\n")
        yield conn.write(data.encode('utf-8'))
        res = yield conn.read_until(b'\r\n\r\n')
        print res

        self.termin_conn = conn

        @tornado.gen.coroutine
        def test():
            while True:
                try:
                    data = yield conn.read_bytes(1024, partial=True)
                    self.write_message(data)
                except tornado.iostream.StreamClosedError:
                    self.close()
                    break

        yield test()

    @tornado.gen.coroutine
    def on_message(self, message):
        try:
            yield self.termin_conn.write(message.encode('utf-8'))
        except tornado.iostream.StreamClosedError:
            self.write_message('Terminal has disconnected.')
            self.close()

    def on_close(self):
        try:
            # 这里可能有暗坑，比如进入 vim 后，页面被关闭（WebSocket 关闭）
            # 如果发送 exit 到容器里的话，是没法退出的
            self.termin_conn.write('\nexit\n') # exit bash when ws close
            self.termin_conn.close()
        except tornado.iostream.StreamClosedError:
            pass
        print "close"

    def check_origin(self, origin):
        # just for test
        return True

Tornado 是我非常喜欢的一个框架，但是它缺失了很多功能模块，比如说 Session，正因为它啥都没有，所以我就爱上它了，这样可以方便自己撸轮子。:D

• Session 原理
• 为 Tornado 添加 Session
  • SessionMixin
  • 结合 Tornado
  • Session Storage
• 感谢

Session 是由于 HTTP 协议是无状态的，所以需要一种机制来保持客户端和服务器之间的会话。

HTTP 协议的流程是

Client -> Send Request -> Server -> Return Response

上述过程，是一次性的，也就是说当客户端发送请求直到服务器返回响应之后，客户端和服务器之间就再也没有任何联系了。于是一般人都会想到，在每次请求时带上一个特定的标识符，然后服务器端记录下来，每次根据特定的标志符就可以知道客户端的身份了。那么，这个特定的标志符就是 session id。这个就是大致上 session 的原理了。

一般情况下，session id 是储存在 cookie 里，并且通过 HTTP Header 传递给服务器，当然，也可以作为 HTTP Query parameters，只是这样个人觉得不太优雅，当然也有一些不安全的因素存在。

def open_session(self):
    self._session_name = self.get_secure_cookie('tornado-session')
    self.session = self.session_storage.get(self._session_name) or {}
    self._session_age = 3600 * 24 * 31

def set_session_age(self, expires):
    self._session_age = expires

def save_session(self):
    if not getattr(self, '_session_name'):
        self._session_name = str(uuid.uuid4())
        self.set_secure_cookie('wing2-session', self._session_name)
    self.session_storage.set(self._session_name, self.session, self._session_age)

<div id="with-tornado"></div>
## 结合 Tornado
有了 `SessionMixin` 之后，就是结合 Tornado 的使用了。
``` python
class BaseHandler(tornado.web.RequestHandler, SessionMixin):
    session_storage = dict()

    def prepare(self):
        self.open_session()

        if self.session.get('user_id'):
            user_id = self.session['user_id']
            self.current_user = UserModel.get_by_user_id(user_id)

    def finish(self, chunk=None):

        self.save_session()
        super(BaseHandler, self).finish(chunk)

上面是初始化 Session，然后继承 BaseHandler，并且在 RequestHandler 里使用 self.session 就可以操作 Session 了。

首先是一个抽象的 Session Storage 的类

class SessionStorage(object):

    def get(self, key, default=None):
        raise NotImplemented

    def set(self, key, value, expires=None):
        raise  NotImplemented

    def delete(self, key):
        raise NotImplemented

基本实现上述 3 个方法，就可以成为一个 Session 的容器了。

以下是 Redis 作为容器的示例

class RedisStorage(SessionStorage):

    def __init__(self, redis, prefix="wing2"):
        self._redis = redis
        self._prefix = prefix

    def _wrapper(self, key):
        return "session:{prefix}:{key}".format(
            prefix=self._prefix,
            key=key
        )

    def get(self, key, default=None):
        key = self._wrapper(key)
        value = self._redis.get(key)
        if not value:
            return default
        value = pickle.loads(value)
        return value

    def set(self, key, value, expires=None):
        key = self._wrapper(key)
        value = pickle.dumps(value)

        return self._redis.setex(key, value, expires or 86400)  # default one day

    def delete(self, key):
        key = self._wrapper(key)
        return self._redis.delete(key)

以上就是为 Tornado 添加 Session 的示例。

EOF

最近学习了下用户权限划分的数据库结构，并且结合到了 Flask 和 SQLAlchemy 中

• 基础表
  • 用户表
  • 角色表
  • 权限表
  • 菜单表
• 关联表
  • 用户角色表
  • 角色权限表
  • 角色菜单表
• SQLAlchemy
• 与 Flask 结合

首先是数据库的整体结构图（简化版）

class UserModel(db.Model):
    __tablename__ = 'user'
    username = db.Column(db.String(50))
    password = db.Column(db.String(128))
    email = db.Column(db.String(128))
    mobile = db.Column(db.String(11))
    name = db.Column(db.String(50))
    gender = db.Column(db.SmallInteger)  # 0 未知， 1 男 2 女

class RoleModel(db.Model):
    __tablename__ = 'role'
    name = db.Column(db.String(20))

class PermissionModel(db.Model):
    __tablename__ = 'permission'
    name = db.Column(db.String(50))
    action = db.Column(db.String(250), unique=True)

class MenuModel(db.Model):
    __tablename__ = 'menu'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50))
    icon = db.Column(db.String(50))
    url = db.Column(db.String(250))
    order = db.Column(db.SmallInteger, default=0)
    bg_color = db.Column(db.String(50))

基础表完了就是关联表了

用户跟角色，肯定是多对多的关系，按照 Flask-SQLAlchemy 里的 Many-to-Many Relationships

user_role = db.Table('user_role',  # 用户角色关联表
    db.Column('user_id', db.Integer, db.ForeignKey('user.id')),
    db.Column('role_id', db.Integer, db.ForeignKey('role.id')),
    db.Column('created_at', db.DateTime, default=datetime.now),
)

这里把权限挂在了角色下面，其实也可以去掉角色，直接跟用户挂钩，但是如果后期在后台分配用户权限，估计会累死。这里角色跟权限也是多对多

role_permission = db.Table('role_permission',  # 角色权限关联表
    db.Column('permission_id', db.Integer, db.ForeignKey('permission.id')),
    db.Column('role_id', db.Integer, db.ForeignKey('role.id')),
    db.Column('created_at', db.DateTime, default=datetime.now),
)

同上，也是多对多

role_menu = db.Table('role_menu',  # 用户菜单关联表
    db.Column('role_id', db.Integer, db.ForeignKey('role.id')),
    db.Column('menu_id', db.Integer, db.ForeignKey('menu.id')),
    db.Column('created_at', db.DateTime, default=datetime.now),
    db.Column('is_delete', db.Boolean, default=False),
)

SQLAlchemy

如果需要获取一个用户的角色，可以利用relationship，关联到角色表上

class UserModel(db.Model):
    # ...
    roles = db.relationship(
        'RoleModel',
        secondary=user_role,
        backref=db.backref(
            'users',
            lazy='dynamic'
        )
    )

获取用户的所有权限可以用property

class UserModel(db.Model):
    # ...
    @property
    def permissions(self):
        permissions = PermissionModel.query.join(role_permission).join(RoleModel).join(user_role).join(UserModel).\
            filter(
            UserModel.id == self.id
        )
        return permissions

同理菜单

class UserModel(db.Model):
    # ...
    @property
    def menus(self):
        menus = MenuModel.query.join(role_menu).join(RoleModel).join(user_role).join(UserModel).\
            filter(
            UserModel.id == self.id
        ).order_by(MenuModel.type_, MenuModel.order).all()
        return menus

这样就可以用user.permissions和user.menus来获得用户的权限和菜单了

数据库表结构设计好了，下面就是跟 Flask 的结合了

在 Python 中，用 decorator 可以用来做用户验证，比如下面

def auth(method):
    @functools.wraps(method)
    def wrapper(*args, **kwargs):
        user_id = session.get('user_id')
        if not user_id:
            return abort(403)
        return method(*args, **kwargs)
    return wrapper

@app.router('/user/info')
@auth
def user_info():
    return render_template('user/info.html')

上面就是利用 Python 的 decorator 来认证用户，其实也是简单的权限划分

因为在 Flask 中，每个 view 就是一个函数，所以在权限表中，用action来表示每个 view 的函数名，那么每个 view 就是一个最小的权限单位，如果一个角色拥有这个权限，那么他就可以请求这个 view 的操作。所以可以这样验证权限

class UserModel(db.Model):
    # ...
    def check(self, action):
        permission = self.permissions.filter(PermissionModel.action == action).first()
        return bool(permissions)

然后把这个权限验证写到 decorator 里去

permissions = list()

class Permission(object):

    def __init__(self, module=None, action=None):
        self.module = module
        self.action = action

    def check(self, module, func):
        if not self.current_user:
            return False
        return self.current_user.check('{module}.{action}'.format(
            module=module,
            action=func
        ))

    def deny(self):
        return fail(4003, u'无权访问')

    def __call__(self, func):
        permissions.append({
            'action': '{}.{}'.format(func.__module__, func.__name__),
            'name': func.__doc__
        })
        @wraps(func)
        def decorator(*args, **kwargs):
            if not self.check(func.__module__, func.__name__):
                return self.deny()
            return func(*args, **kwargs)
        return decorator

    def __enter__(self):
        if not self.check(self.module, self.action):
            try:
                self.deny()
            except Exception as e:
                raise e
            else:
                raise PermissionDeniedException

    def __exit(self):
        pass

    @property
    def current_user(self):
        return g.user

permission = Permission()

这里参考了 hustazp 的 permission

使用 func.__module__ 和 func.__name__ 结合作为权限中的 action，如果单独用 func.__name，肯定会出现相同的函数名，如果加上 func.__module__ 就在一定程度上避免了重合，并且将 func.__doc__ 来作为权限种的 name，还没想到更好的办法来自动加入 name。

那么上面的用户认证换成 permission 就是下面

@app.router('/user/info')
@permission
def user_info():
    """用户信息"""
    return render_template('user/info.html')

在开发的过程中，如果写了一个权限就要加到数据库里该有多累，于是就加了一个 permissions，这里把所有的 view 都加到这里面来，然后通过下面的脚本来加入权限

from application.models.user import PermissionModel, RoleModel, role_permission
from application.utils.permissions import permissions
for permission in permissions:
    p = PermissionModel.query.filter_by(action=permission['action']).first()
    if not p:
        p = PermissionModel(
            name=permission['name'],
            action=permission['action']
        )
        db.session.add(p)
        db.session.commit()

role = RoleModel.query.first()  # 这里默认获取一个角色，并且赋予权限
for p in PermissionModel.query.filter_by(is_delete=False):
    r = db.session.query(role_permission).join(RoleModel).join(PermissionModel).\
        filter(
            RoleModel.id == role.id,
            RoleModel.is_delete == False,
            PermissionModel.id == p.id,
            PermissionModel.is_delete == False,
            role_permission.c.is_delete == False,
        ).first()
    if not r:
        role.permissions.append(p)

role.save()

Tornado 是一个 Python 的 Web 框架和一个异步网络库。

Tornado is a Python web framework and asynchronous networking library

单纯作 Python 的 Web 框架，Tornado 并没有啥特色，和 Web.py 的接口类似。话说我学的第一个 Python Web 框架就是 Web.py，导致我对相似的 Tornado 恋恋不忘。于是乎，我就开始阅读 Tornado 的源代码了。

就像上面的介绍，Tornado 除了是一个 Web 框架，还是一个异步的网络库。而这个库的核心和灵魂就是 torndo.ioloop 了。

Content

• Content
• socket
  • server
  • client
  • blocking network
• socket based on threading
• select
• epoll
• tornado.ioloop
• refs

socket

作为网络库，就不得不从 socket 说起了。
虽然也是读过大学的人，但是当时网络、操作系统完全等于没学过，只怪自己不努力，所以现在对于 socket，只能从应用的层面去学习了。

server

import socket

s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

s.bind(('127.0.0.1', 5000))
s.listen(1)
print "waiting for connect"
conn, addr = s.accept()
print conn, addr

while True:
    data = conn.recv(1024)
    if len(data) == 1:
        if ord(data) == 4:
            conn.sendall('bye.\n')
            break
    conn.sendall('Hello\n')

s.close()

首先是通过socket.socket创建一个socket对象，第一参数有三个可选

• socket.AF_UNIX UNIX socket file，但是要系统支持
• socket.AF_INET IPv4 地址
• socket.AF_INET6 IPv6 地址

第二个参数，指定 socket 类型，有很多类型可选，但是常用的是下面两个

• socket.SOCK_STREAM TCP 协议
• socket.SOCK_DGRAM UDP 协议

然后通过s.bind来绑定socket监听的地址和端口，如果需要让所有地址都能连接上，可以指定0.0.0.0

最后通过s.accept开始等待客户端连接上，这里是阻塞的，意思就是如果没有客户端连接上，后面的代码是永远也不会执行的。

下面通过telnet 127.0.0.1 5000来测试上面的效果

> telnet 127.0.0.1 5000
Trying 127.0.0.1...
Connected to localhost.
Escape character is '^]'.
hello
Hello
hi
Hello
bye.
Connection closed by foreign host.

当连接telnet连接上 server 时，代码就会进入循环中，等待客户端发送数据，当客户端发送了数据之后，server 就会向客户端发送hello，这里的s.recv也是阻塞的。

最后如果收到了客户端发来的EOF（C-D）的时候，就会退出循环，然后通过s.close关闭 socket 后退出程序。

client

上面是通过telnet作为客户端来测试，下面通过 Python 里的 socket 来建立链接

import socket

s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

s.connect(('127.0.0.1', 5000))
s.sendall('hello')
data = s.recv(1024)
s.sendall(chr(4))
s.close()
print data

跟 server 一样，首先是初始化socket对象，然后通过s.connect连接到 server，接下来通过s.sendall发送数据到 server，这里的s.connect和s.recv也是阻塞的。

blocking network

上面的 server 和 client 都是阻塞的，或者说 server 只能接受一个 client 的连接，client 也只能连接到一个 server 上去。

但是实际情况比如一个网站（web server）是可以同时多个人浏览，web server 是 HTTP 协议，而 HTTP 协议是基于 TCP（socket）的，并且不止是接受一个连接。

socket based on threading

下面是 server 的例子

import socket
import threading

s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

s.bind(('127.0.0.1', 5000))
s.listen(5)
print "waiting for connect"


class Server(threading.Thread):

    def __init__(self, sock):
        self.sock = sock
        super(Server, self).__init__()

    def run(self):
        conn, addr = self.sock.accept()
        print "client {} joined\n".format(addr)
        while True:
            data = conn.recv(1024)
            if len(data) == 1:
                if ord(data) == 4:
                    break
            print "data {} from {}".format(data, addr)
            conn.sendall('Hello\n')
        self.sock.close()
        print "client {} left\n".format(addr)

thread = list()
for i in xrange(5):
    t = Server(s)
    thread.append(t)
    t.start()

通过threading来启动了5个socket.accept来接收 client 的连接，通过telnet可以同时启动5个 client，并且可以同时工作，如果需要接受更多的 client，可以增加线程数，不过增加线程带来的结果就是内存飚升，会影响性能，据说 windows 下有限制子线程数，也就是限制了客户端的连接数。

select

为了解决阻塞的问题，牛逼的人类发明了select这个系统调用，多路 I/O 复用，可以在单线程里同时处理多个 socket 连接。在类 Unix 系统中，网络被抽象成了文件，可读可写，所以作为一个文件描述符，可以被select这个系统调用监视。

import socket
import select

s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

s.bind(('127.0.0.1', 5000))
s.listen(5)
rlist = [s]
wlist = []
message = dict()

print "waiting connect"
while rlist:
    readable, writable, exceptional = select.select(rlist, wlist, rlist)
    for sock in readable:
        if sock is s:
            conn, addr = s.accept()
            print "client {} joined.".format(addr)
            message[conn] = list()
            rlist.append(conn)
        else:  # conn
            data = sock.recv(1024)
            if len(data) == 1:
                if ord(data) == 4:
                    sock.close()
                    if sock in wlist:
                        wlist.remove(sock)
                    if sock in rlist:
                        rlist.remove(sock)
                    if sock in message:
                        del message[sock]
                    print "client {} left.".format(addr)
                    continue
            message[sock].append('Hello\n')
            if sock not in wlist:
                wlist.append(sock)

    for sock in writable:
        if sock in message:
            if message[sock]:
                data = message[sock].pop()
                if data:
                    sock.sendall(data)

    for sock in exceptional:
        rlist.remove(sock)
        print "error occur."
        sock.close()
        del message[sock]

上面代码看着变得复杂起来了，其实细细看来还是很好理解的，select.select接受三个参数，

• rlist 可读的文件描述符
• wlist 可写的文件描述符
• xlist 异常的对象

并且select.select 会阻塞，直到有可操作的对象（文件描述符），然后返回readable, writable, exceptional，然后通过循环遍历这三个对象（文件描述符数组）进行操作。

在readable循环里，如果可读的对象是当前socket对象，说明有新的连接进入了，通过sock.accept得到的连接对象，这个连接对象也是一个可读可写的文件描述符，所以要得到新的连接的数据，将这个对象也放入到rlist里面去。当readable循环里的对象不是当前socket对时，那肯定就是建立的连接有了客户端发来的数据了，通过sock.recv就可以得到数据了。同理writable。

使用select，其中需要监视的文件描述符随着连接的加入，会不断的变多，因为select是遍历的数组，数组的长度是有限制的，并且对于不活跃的文件描述符，select还是会不停的遍历，这样效率极低。

epoll

由于select会有各种限制，所有在 Linux 中，又出现了epoll。

import socket
import select

s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

s.bind(('0.0.0.0', 5000))
s.listen(1)
s.setblocking(0)
print "waiting for connect"

epoll = select.epoll()
epoll.register(s.fileno(), select.EPOLLIN)


connections = {}
requests = {}
responses = {}
clients = {}
try:
    while True:
        events = epoll.poll(1)
        for fileno, event in events:
            if fileno == s.fileno():  # sock
                conn, addr = s.accept()
                conn.setblocking(0)
                epoll.register(conn.fileno(), select.EPOLLIN)  # add conn to read
                connections[conn.fileno()] = conn
                clients[conn.fileno()] = addr
                print "client {} joined".format(addr)
            elif event & select.EPOLLIN:
                data = connections[fileno].recv(1024)
                print "recv data {} from client {}".format(data, clients[fileno])
                if len(data) == 1:
                    if ord(data) == 4:
                        responses[fileno] = "bye\n"
                else:
                    responses[fileno] = "hello\n"
                epoll.modify(fileno, select.EPOLLOUT)
            elif event & select.EPOLLOUT:
                if fileno in responses:
                    try:
                        connections[fileno].sendall(responses[fileno])
                        if responses[fileno] == "bye\n":
                            epoll.modify(fileno, 0)
                            connections[fileno].shutdown(socket.SHUT_RDWR)
                        else:
                            epoll.modify(fileno, select.EPOLLIN)
                        del responses[fileno]
                    except:
                        epoll.modify(fileno, 0)
            elif event & select.EPOLLHUP:
                epoll.unregister(fileno)
                connections[fileno].close()
                del connections[fileno]
                print "client {} left".format(clients[fileno])
                del clients[fileno]
finally:
    epoll.unregister(s.fileno())
    epoll.close()
    s.close()

上面无限循环之前，是常规的 socket 对象的创建，然后是epoll对象的创建和 socket 对象的文件描述符的fileno作为EPOLLIN事件加入到epoll里去监视。

然后就进入了无限次的循环中，然后通过epoll.poll系统调用，获取可操作的文件描述符，和对应的事件，因为只注册了一个EPOLLIN事件，所以当有 client 要连接的时候，就可以通过socket.accpet来建立连接，然后再把建立的连接（也是一个文件描述符）加入到epoll中去监视读数据。当有EPOLLIN进入时，就可以从可读连接里去读数据了，在这里我设定如果读到数据就给 client 发送hello，所以就把当前连接（文件描述符）修改为EPOLLOUT事件，然后在进入EPOLLOUT事件时，就把数据发送给客户端去。

那么 epoll不会不停的扫描所有的文件描述符，而是在有事件发生时，才会处理这个事件。

tornado.ioloop

为什么看tornado.ioloop会有上面这些，因为特么的直接看tornado.ioloop，我看不太懂，经过上面一步一步下来，对于理解tornado.ioloop里的代码有很大的帮助。

因为是以学习为目的，所以从 Tornado 的v1.2这个版本的tornado.ioloop来看，因为少，才557行，所以荣誉理解。

tornado.ioloop是以epoll为基础的，但是在不支持epoll的平台，会选择该平台对应的实现来做。

从这里开始

# Choose a poll implementation. Use epoll if it is available, fall back to
# select() for non-Linux platforms
if hasattr(select, "epoll"):
    # Python 2.6+ on Linux
    _poll = select.epoll
elif hasattr(select, "kqueue"):
    # Python 2.6+ on BSD or Mac
    _poll = _KQueue
else:
    try:
        # Linux systems with our C module installed
        import epoll
        _poll = _EPoll
    except:
        # All other systems
        import sys
        if "linux" in sys.platform:
            logging.warning("epoll module not found; using select()")
        _poll = _Select

如果是 Python 2.6+ 在 Linux 上，选择select下的 epoll，如果是在 BSD 或者 Mac，则选择kqueue，如果没有则首先尝试 Tronado 写的 C 扩展的epoll模块，最后实在不行，就使用select。

Tronado 将不同的平台不同的 poll 实现了相同的接口，_Select、_KQueue、_EPoll，方便在IOLoop里使用。

在IOLoop

• add_handler相当于 epoll.register
• update_handler相当于epoll.modify
• remove_handler相当于epoll.unregister

然后就是IOLoop.start了，也是一个无限次循环，然后抛弃其他的不看，直接进入243行的event_pairs = self._impl.poll(poll_timeout)，然后是267行这里，得到文件描述符fd和events，接下来就像上述的epoll操作一样了，只是tornado.ioloop增加了一些 helper ，方便了操作而已。简化下来就是下述的代码了

class IOLoop(object):

    def __init__(self, impl=None):
        self._impl = impl or _poll()

    def add_handler(self, fd, handler, events):
        self._impl.register(fd, events | self.ERROR)

    def update_handler(self, fd, events):
        self._impl.modify(fd, events | self.ERROR)

    def remove_handler(self, fd):
        self._impl.unregister(fd)

    def start(self):
        while True:
            event_pairs = self._impl.poll(poll_timeout)

            self._events.update(event_pairs)
            while self._events:
                fd, events = self._events.popitem()
                self._handlers[fd](fd, events)

上面就是极其简化后的IOLoop了，看起来是不是跟epoll很像啊。

然后结合tornado.ioloop的示例代码

import errno
import functools
import ioloop
import socket
def connection_ready(sock, fd, events):
    while True:
        try:
            connection, address = sock.accept()
        except socket.error, e:
            if e.args[0] not in (errno.EWOULDBLOCK, errno.EAGAIN):
                raise
            return
        connection.setblocking(0)
        handle_connection(connection, address)
        
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.setblocking(0)
sock.bind(("", port))
sock.listen(128)
io_loop = ioloop.IOLoop.instance()
callback = functools.partial(connection_ready, sock)
io_loop.add_handler(sock.fileno(), callback, io_loop.READ)
io_loop.start()

在之前，完全是不懂上面的示例代码的意思，经过层层剥茧，从最原始的 socket 开始理解，就很好的理解了tornado.ioloop对于网络的处理了。

refs

• https://docs.python.org/2/library/select.html
• https://docs.python.org/2/library/threading.html
• https://pymotw.com/2/select/
• https://github.com/tornadoweb/tornado/blob/branch1.2/tornado/ioloop.py

自从在上海吃过麻辣香锅之后，从此爱上了这个，原谅我这土包子没吃过好的。

昨日在张大妈白菜党看到了二手东海底捞麻辣香锅特价，19.04RMB，加上我二手东钻石会员，每个月有2张免运费券，用也用不完。果断下单入手，于是开始了下面的麻辣香锅之旅。

食材采购

首先就是选购食材了，根据在知乎和下厨房看到的教程，以及平时在外面店里吃的结合起来，列出了如下菜单（只有做这种事，才想到了印象笔记的好处，买一个勾一个）

• 土豆
• 藕
• 笋尖
• 里脊肉
• 鸡胸肉
• 鸡翅
• 骨肉相连
• 红薯
• 牛肉
• 腐竹
• 黑木耳
• 茶树菇
• 香菇
• 金针菇
• 火腿
• 虾
• 葱
• 五花肉
• 培根
• 大葱
• 郫县豆瓣酱
• 牛蛙

最终根据实际情况，删除掉的都是没有买（比如难得处理虾）或者因为各种情况买不到（比如牛蛙）的。

食材准备

对于素菜，切丁切片洗净准备，对于鸡翅、鸡胸肉等冷冻食品，放入热水解冻，然后用盐和料酒腌制（如果喜欢清淡，这里要少放盐），最后加入淀粉裹匀。

根据上述知乎与下厨房教程的结合，做出了如下的选择

• 素菜类全部用水煮
• 荤菜一律油炸

经过水煮、油炸之后，食物大概8、9、10成熟的样子。然后锅里放入油（这里的油可以用上述油炸荤菜剩的油）、葱、大葱爆炒，然后加入二手东买的佐料，『炒出香味后』（来自说明书），加入食材后，『加入少量水收汁』（也是来自说明书），然后不停翻炒，直至汁水木有了。

由于家中锅太小，盆太少，最终分了4次爆炒，最终出了一大锅。

最终效果图

最终耗时2小时，终于完成了如下的美味（海底捞佐料真的很美味）。

最近搬了新家，4个人合租，办了上海电信的 30MB 的光纤，说是 30MB，上行才 2MB，好在下行是实打实的 30MB，于是就准备把吃灰的树莓派找出来继续做下载机下电影美剧看。

之前在寝室住的时候，直接在路由器上拨号上网，有公网 IP，用的路由器可以做端口转发，配合 DDNS，可以很方便的从外面访问控制树莓派，但是现在这边搬的光纤，必须用电信『强制』租给我们一个光猫，押金200元，说好一年之后可以凭光猫退200块，但是之前来给我们办理网络的师傅说甭指望了，退不了的，我直接告诉他，退不了就去工信部投诉，合同上写好了可以退（其实我也没看合同上到底有没有写，不是我去办理的）。『强租』也就算了，拨号的账号密码都写在光猫里，不给我们，超级管理员密码也不知道，问电信的师傅，说他也不知道，导致我们不能用自己的光猫，这也引发了此片文章。

通过curl ip.cn看了下，电信还算良心给分配了公网 IP，然后电信『强租』的光猫还是一个路由器，有普通管理员账号密码，进去后发现有 DMZ 和虚拟主机配置（端口转发），然后尝试了配置 DMZ 和虚拟主机配置（端口转发），结果发现怎么都不生效，于是不得不放弃之。然后 Google 之，发现了关键字『SSH 反向代理』。

然后按照这里的配置进行配置，然而配置好了后，SSH 连接时，会给出下面错误

channel 2: open failed: connect failed: Connection refused

又一次 Google 之，发现了这个，然而虽然没有解决我的问题，但是通过这个回答，我试了下把端口统一下

ssh -fCNR <port_b1>:localhost:22 usr_b@B.B.B.B
ssh -fCNL "*:<port_b1>:localhost:<port_b1>' localhost

注意上述，按照这里的配置，应该为

ssh -fCNR <port_b1>:localhost:22 usr_b@B.B.B.B
ssh -fCNL "*:<port_b2>:localhost:<port_b1>' localhost

当我按照这个配置，得到了错误，我就试了下把端口统一下，然后竟然成功了。但是按照文档，port_b2可以与port_b1不同，但我就是不能成功。

上述配置成功了，就可以从外面登陆到树莓派了

ssh -p <portb1> usra@B.B.B.B

然而，因为网络的原因，ssh会断开，并且不会重连，虽然可以通过修改sshd_config配置修改连接时间，但是还是会断开。于是我就用supervisor来是ssh一直连接着。

这里需要注意以下几点：

• 我去掉了-f参数，如果加上-f，supervisor就不能正常的监控ssh了；
• ssh我双向配置了 key 登陆，所以就不用输入密码，supervisor可以自动启动ssh；
• 因为配置了 key 登陆，而我的supervisor是以root用户运行，所以需要在supervisor里指定user参数，使ssh可以正常读取到id_rsa，当然也可以用-i参数显示的指定id_rsa路径。

update

2016-06-02
对于 ssh 的自动重连，可以使用 autossh 配合 supervisor

并且 ssh 应该加上下列参数

-o "ServerAliveInterval 60" -o "ServerAliveCountMax 3" -o "StrictHostKeyChecking no"

不然，autossh 会『假死』

ServerAliveInterval的作用是当没有数据交互时，间隔一段时间给服务器发送一个message，并且接受服务器的响应

ServerAliveCountMax的作用是ServerAliveInterval最大重试次数，当失败了就会中断连接，这样 autossh 又能自动重连了

完整的的命令是

autossh -M 0 -q -o "ServerAliveInterval 60" -o "ServerAliveCountMax 3" -o "StrictHostKeyChecking no" -CNR 2222:127.0.0.1:22 -i .ssh/id_rsa user@host

本机连接到 host主机，并且在 host 上设置一个 127.0.0.1:2222的反向代理到本机

扇贝打卡只有分享到微博、微信、QQ，然而还要手动分享，这样不利于监督。
于是我就想利用 IFTTT 来实现自动化操作，因为 IFTTT 并没有提供扇贝的 channel，所以曲线救国，就考虑到了通过 RSS，然后扇贝也没有提供打卡的 RSS 源，于是就只能手动生成 RSS 源了。

在很久很久以前，我记得扇贝有提供一个公开的个人打卡记录的页面，貌似是这个 http://shanbay.com/checkin/user/4539052/ ，但是现在这个 URL 必须登录后才能访问，本来想着模拟登录的，但是发现打卡的公共排行榜可以访问，例如『最受欢迎打卡』，那就说明还是有办法可以不登陆访问个人的打卡记录的嘛，于是辗转曲折的找到了『XX 最近打卡日记』，例如 http://shanbay.com/checkin/record/25683094549/ 这个，通过这个页面，有最近的7条打卡记录，如果是做自动同步，7条记录就够了，然后就有了下面的程序了。

http://shanbay2rss.hui.lu/shanbay2rss/<your_record_id>，这是一个可用的扇贝打卡 RSS 源。