Kubernetes 实战-日志处理

发表于 2019-05-17 |

基础日志处理

在 Kubernetes（简称 k8s）中，所有应用在 Pod（k8s 管理容器最小单位）中运行，标准处理方式为将日志打印到标准日志输出和标准错误输出，这样我们可以通过 kuberctl logs 关键字获取容器运行时日志，根据容器运行时的类型不同，日志保存路径也不同，以 Docker 为例，所有真实日志均在 /var/lib/docker/ 路径下，下面我们来看一个例子：

在 k8s 中创建一个 Pod，Pod 中指定打印当前时间到标准输出中：

apiVersion: v1
kind: Pod
metadata:
  name: counter
spec:
  containers:
  - name: count
    image: busybox
    args: [/bin/sh, -c,
            'i=0; while true; do echo "$i: $(date)"; i=$((i+1)); sleep 1; done']

运行该 Pod，通过 kubectl logs 获取当前 Pod 日志：

[root@node1 blog]# kubectl get pod |grep counter
counter            1/1     Running   0          9s
[root@node1 blog]# kubectl logs counter
0: Fri May 17 00:34:01 UTC 2019
1: Fri May 17 00:34:02 UTC 2019
2: Fri May 17 00:34:03 UTC 2019
3: Fri May 17 00:34:04 UTC 2019
4: Fri May 17 00:34:05 UTC 2019
5: Fri May 17 00:34:06 UTC 2019
6: Fri May 17 00:34:07 UTC 2019
7: Fri May 17 00:34:08 UTC 2019
8: Fri May 17 00:34:09 UTC 2019
9: Fri May 17 00:34:10 UTC 2019
10: Fri May 17 00:34:11 UTC 2019
11: Fri May 17 00:34:12 UTC 2019
12: Fri May 17 00:34:13 UTC 2019

这里如果使用 -f 选项，可以持续输出该 Pod 日志。

那么我们如何找到该容器对应的实际日志文件呢？

我们可以现在 Docker 中找到该容器信息：

1 2	[root@node1 blog]# docker ps \|grep count 012e0352b193 busybox "/bin/sh -c 'i=0; wh…" 2 minutes ago Up 2 minutes k8s_count_counter_default_75792a2a-783b-11e9-a3d9-525400aea01a_0

可以看到在 Docker 中该容器名称为 k8s_count_counter_default_75792a2a-783b-11e9-a3d9-525400aea01a_0 ，我们先不去管最后的 UUID 是什么含义，先看前几个字段，跟集群信息关联可以看到，分别是：k8s，容器名称，Pod 名称，namespaces 名称。那么我们来看下这个容器在 Docker 中的配置文件：

[root@node1 containers]# pwd
/var/lib/docker/containers
[root@node1 containers]# ls
012e0352b19387170b903aff7c73c02fcd023c2f53cbe5b908392ff4e1a126e2  
...
5c9e765f68a15d5510870179160c59d45d0287b9f92265687d6c37a3557a1017  c563c36b0d3d28ea611b270bc1609ae9c0e720c1c1f85ba669cdab5f7099b77b

在 Docker 容器路径下，我们看到了很多以不知道什么 ID 明明的子目录，最开始以为是我们 docker ps 中看到的最后 uuid，但是发现对不上，那么我们可以查看下 Pod 的详细信息来试图获取:

[root@node1 containers]# kubectl describe pod counter
Name:         counter
Namespace:    default
Node:         192.168.27.231/192.168.27.231
Start Time:   Fri, 17 May 2019 08:33:58 +0800
Labels:       <none>
Annotations:  <none>
Status:       Running
IP:           172.20.2.13
Containers:
  count:
    Container ID:  docker://012e0352b19387170b903aff7c73c02fcd023c2f53cbe5b908392ff4e1a126e2
    Image:         busybox
    Image ID:      docker-pullable://busybox@sha256:4b6ad3a68d34da29bf7c8ccb5d355ba8b4babcad1f99798204e7abb43e54ee3d

忽略掉无关信息，我们可以看到 Container ID 字段，这里对应的 ID 就是在 /var/lib/docker/containers/ 下的 ID，找到了 ID，我们可以来看看具体的配置：

[root@node1 012e0352b19387170b903aff7c73c02fcd023c2f53cbe5b908392ff4e1a126e2]# tree .
.
├── 012e0352b19387170b903aff7c73c02fcd023c2f53cbe5b908392ff4e1a126e2-json.log
├── checkpoints
├── config.v2.json
├── hostconfig.json
└── mounts

2 directories, 3 files

可以看到有配置文件和日志文件，我们今天只来讨论日志，那么我们看下这个日志文件的内容：

[root@node1 012e0352b19387170b903aff7c73c02fcd023c2f53cbe5b908392ff4e1a126e2]# tail 012e0352b19387170b903aff7c73c02fcd023c2f53cbe5b908392ff4e1a126e2-json.log
{"log":"706: Fri May 17 00:45:48 UTC 2019\n","stream":"stdout","time":"2019-05-17T00:45:48.913451073Z"}
{"log":"707: Fri May 17 00:45:49 UTC 2019\n","stream":"stdout","time":"2019-05-17T00:45:49.915605471Z"}
{"log":"708: Fri May 17 00:45:50 UTC 2019\n","stream":"stdout","time":"2019-05-17T00:45:50.917823388Z"},
{"log":"713: Fri May 17 00:45:55 UTC 2019\n","stream":"stdout","time":"2019-05-17T00:45:55.928645971Z"}
{"log":"714: Fri May 17 00:45:56 UTC 2019\n","stream":"stdout","time":"2019-05-17T00:45:56.930660054Z"}
{"log":"715: Fri May 17 00:45:57 UTC 2019\n","stream":"stdout","time":"2019-05-17T00:45:57.932763561Z"}

可以看到这里日志输出是 JSON 格式的，这里跟 kubectl logs 输出不一致啊，为啥这里是 JSON 呢？其实这跟你的 Docker 配置有关，具体的配置可以在 Docker 配置中看到，比如：

[root@node1 docker]# pwd
/etc/docker
[root@node1 docker]# cat daemon.json
{
  "registry-mirrors": [
    "https://dockerhub.azk8s.cn",
    "https://docker.mirrors.ustc.edu.cn",
    "http://hub-mirror.c.163.com"
  ],
  "max-concurrent-downloads": 10,
  "log-driver": "json-file",
  "log-level": "warn",
  "log-opts": {
    "max-size": "10m",
    "max-file": "3"
    },
  "data-root": "/var/lib/docker"
  }

Docker 默认的日志格式是 JSON，这里我猜测是 k8s 通过 Docker 接口获取日志类型，然后进行相应的解析输出（希望之后对 k8s 更深入的了解来验证猜想）。

在了解了标准日志处理，那么我们来看下节点级别的日志处理是怎样的。

节点级别日志

标准处理

我们现在知道了 Pod 的日志其实是存放在容器真正运行所在节点上的，那么如果 Pod 一直运行，日志会不断增大，占用很多的日志空间，这个在节点上是怎么控制的呢？

因为这种方式不是 k8s 推荐的方式，这里并没有采用集群级别的控制方式，而是以节点为粒度的，各个节点通过 logrotate 自己处理日志轮询，logrotate 相信大部分同学都使用过，这里不详细说了。

特殊处理

如果我们不想将应用日志都输出到标准输出，想将日志打印到 /var/log/ 下的自定义路径下怎么办？我们可以在 Pod 启动时挂载一个 Volume，这个 Volume 就是Pod 所在的节点真实路径，这样我们在容器中就可以直接将日志打印到该路径下，哪怕 Pod 被销毁，日志也会一直存在。

我们创建一个 Deployment 类型的资源，在创建之前，我们需要先创建出指定的挂载路径: /var/log/yiran/，资源配置如下：

[root@node1 blog]# cat log.yaml
apiVersion: apps/v1beta1
kind: Deployment
metadata:
  name: counter
spec:
  replicas: 1
  template:
    metadata:
      labels:
        run: helloworldanilhostpath
    spec:
      containers:
      - name: count
        image: busybox
        args: [/bin/sh, -c,
              'i=0; while true; do echo "$i: $(date)" >> /var/log/yiran/test.log; i=$((i+1)); sleep 1; done']
        volumeMounts:
        - name: yiran-test-log
          mountPath: /var/log/yiran
      volumes:
      - name: yiran-test-log
        hostPath:
          path: /var/log/yiran
          type: Directory

创建完成后，我们来查看下该 Pod 的日志：

[root@node1 blog]# kubectl get pod
NAME                       READY   STATUS    RESTARTS   AGE
counter-76b584fd8f-7fq99   1/1     Running   0          2m30s
testlog-rs-lwxts           1/1     Running   0          2d1h
[root@node1 blog]# kubectl logs counter-76b584fd8f-7fq99

可以看到没有标准日志输出，我们按照上面描述，去看下容器对应的 Docker 日志是否为空：

[root@node2 166103275a89df918b3240ede911192357f7256c1f23b4311b6db44c0f800cc2]# pwd
/var/lib/docker/containers/166103275a89df918b3240ede911192357f7256c1f23b4311b6db44c0f800cc2
[root@node2 166103275a89df918b3240ede911192357f7256c1f23b4311b6db44c0f800cc2]# ll
total 12
-rw-r-----. 1 root root    0 May 18 17:39 166103275a89df918b3240ede911192357f7256c1f23b4311b6db44c0f800cc2-json.log
drwx------. 2 root root    6 May 18 17:39 checkpoints
-rw-------. 1 root root 5173 May 18 17:39 config.v2.json
-rw-r--r--. 1 root root 2064 May 18 17:39 hostconfig.json
drwx------. 2 root root    6 May 18 17:39 mounts

这里是符合预期的，因为我们将所有的日志输出到指定日志： /var/log/yiran/test.log 中了，我们去看看节点日志是否存在：

[root@node2 yiran]# pwd
/var/log/yiran
[root@node2 yiran]# tailf test.log
327: Sat May 18 09:45:23 UTC 2019
328: Sat May 18 09:45:24 UTC 2019
329: Sat May 18 09:45:25 UTC 2019
330: Sat May 18 09:45:26 UTC 2019

可以看到这里已经按照预期打印在 Pod 所在节点上了，满足我们的需求。

接下来是重头戏，集群级别的日志控制。

集群级别日志

我们为什么需要日志？对于开发者，可以通过日志进行快速的错误排查；对于运维同学可以根据日志来了解程序运行状态。一句话就是日志非常重要。

那么既然日志这么重要，那么在 k8s 上如何处理日志，尤其是 k8s 提供了 ReplicaSet/DeploymentSet 这类可以自动缩扩容，自动 ha 的资源，我们如果仅仅通过节点级别的日志管理，集群规模小还好，当集群规模变大之后，对于使用日志的同学简直是灾难。

在看过了标准日志处理和节点日志处理后，我们来看看一个标准的 k8s 集群是如何处理服务日志和应用日志的。

这里的前提是，我们有一个日志中心（如 ES)去处理日志，有几种配置方式可以选择：

节点级别日志代理

在 k8s 集群中运行一个 DaemonSet，启动的容器运行日志转发器，用于将节点上的日志转发到日志中心，日志转发器可以根据各自资源情况和需求自由选择，如 Logstash,Fluentd,Fluent-bit 等等。

k8s 标准配置中推荐该方案，无论是从资源使用还是从配置管理上都是最佳方案。

对应配置在 kubernetes/cluster/addons/fluentd-elasticsearch 路径下，我们可以直接创建部署，部署后状态如下：

[root@node1 fluentd-elasticsearch]# kubectl -n kube-system get rs |grep kibana
kibana-logging-f4d99b69f          1         1         1       2d6h
[root@node1 fluentd-elasticsearch]# kubectl -n kube-system get ds |grep fluent
fluentd-es-v2.4.0       3         3         3       3            3           <none>                          2d6h
[root@node1 fluentd-elasticsearch]# kubectl -n kube-system get statefulset
NAME                    READY   AGE
elasticsearch-logging   2/2     2d6h

可以看到 Fluentd 是以 DaemonSet 方式运行的；ElasticSearch 是以 StatefulSet 方式运行的；Kibana 是以 ReplicaSet 方式运行的。

其中 Fluentd 配置文件中监控的是 /var/log/container 路径下的所有日志，所以我们理论上可以在 ES 中看到所有应用的日志，那么看到日志之后，我们如何与实际的应用对应呢？这里可以看下具体示例：

创建 ReplicaSet 类型资源，指定 replica 为 1：

[root@node1 ~]# cat log.yaml
apiVersion: apps/v1beta2
kind: ReplicaSet
metadata:
  name: testlog-rs
spec:
  replicas: 1
  selector:
    matchLabels:
      app: testlog-label
  template:
    metadata:
      labels:
        app: testlog-label
    spec:
      containers:
        - name: testlog-container-name
          image: busybox
          args: [/bin/sh, -c,
                 'i=0; while true; do echo "$i: $(date)"; i=$((i+1)); sleep 1; done']

我们将涉及到名称的字段均加上对应 key，便于在 ES 中查看对应关系，那么接下来看看在 ES 中该容器对应的日志是什么形式的：

根据上述对应关系，哪怕 Pod 重建了，我们仍可以通过 container_name 字段来查看对应应用日志，便于调试。

节点级别日志代理配合伴生容器

我们知道，标准容器日志应该输出到 stdout 和 stderr 中，那么如果我们一个 Pod 中输出多份日志怎么办？虽然这种情况是我们应该极度避免的，我们应该始终保证一个 Pod 只做一件事情。但是我们有时候迫于代码结构或者其他因素，导致我们会遇到这种情况，那么此时我们需要伴生容器配合使用。

示例如下：

admin/logging/two-files-counter-pod-streaming-sidecar.yaml 

apiVersion: v1
kind: Pod
metadata:
  name: counter
spec:
  containers:
  - name: count
    image: busybox
    args:
    - /bin/sh
    - -c
    - >
      i=0;
      while true;
      do
        echo "$i: $(date)" >> /var/log/1.log;
        echo "$(date) INFO $i" >> /var/log/2.log;
        i=$((i+1));
        sleep 1;
      done
    volumeMounts:
    - name: varlog
      mountPath: /var/log
  - name: count-log-1
    image: busybox
    args: [/bin/sh, -c, 'tail -n+1 -f /var/log/1.log']
    volumeMounts:
    - name: varlog
      mountPath: /var/log
  - name: count-log-2
    image: busybox
    args: [/bin/sh, -c, 'tail -n+1 -f /var/log/2.log']
    volumeMounts:
    - name: varlog
      mountPath: /var/log
  volumes:
  - name: varlog
    emptyDir: {}

这种方法是极度不推荐的，如果我们配置了 EFK，那么我们 1 份日志相当于写了 3 份，如果我们 ES 的后端存储是一个副本机制的分布式存储，那么我们 1 份日志相当于写了 3 * 2(或 3，存储副本数)份，这是极大的浪费了存储资源的，且会大大影响 SSD 磁盘寿命。

Pod 级别日志代理

如果觉得节点级别日志代理粒度太粗，我们也可以选择 Pod 级别，在每个 Pod 中都启动一个伴生容器作为日志代理，将日志直接转发到日志中心。

若以该形式部署，则我们的应用程序配置不仅要配置应用自身，还要考虑日志处理策略；节点计算能力现在大幅提升，每个节点的 Pod 数量很大，浪费了大量的计算资源。

应用自处理日志代理

如标题，我们当然可以在应用中直接将日志转发到指定的日志中心，这种情况也是极度糟糕的， Make each program do one thing well.。

总结

到这里我们关于日志部分的介绍就到这里，总的来说我们需要集中式的日志中心，且推荐以节点级别日志代理方式配置，前提是我们能够预留足够的计算资源和存储资源。后续有机会我们可以了解下日志与事件审计关联使用。

下一篇我们来看下 Kubernetes 中镜像相关管理与使用。

Kubernetes 实战-微服务

发表于 2019-05-14 |

微服务

在《Kubernetes In Action》的开始，先要了解 k8s 的需求来自于哪里，为什么我们需要 k8s。

引用维基百科解释：

微服务 (Microservices) 是一种软件架构风格，它是以专注于单一责任与功能的小型功能区块 (Small Building Blocks) 为基础，利用模块化的方式组合出复杂的大型应用程序，各功能区块使用与语言无关 (Language-Independent/Language agnostic) 的 API 集相互通信。

说一说我的理解，在项目早期，都是单体应用，随着功能越来越多，项目越来越大，虽然保证了部署运维的方便，但对于组内同学并不友好，新同学往往要在一坨代码中找自己想要的一点，本地修改提交跑 CI 也是以项目为单位的执行（前段时间 B 站不小心泄露的 Golang 代码就是这种）。当后续升级产品时，因为是以项目为最小粒度，哪怕无关代码，也要被迫进行代码升级，服务重启等操作，带来了额外的风险。

在 2014年，Martin Fowler 与 James Lewis 共同提出了微服务的概念，把单体应用改为通过接口产生的远程方法调用，将项目拆分，一个项目保证只做一件事情，独立部署和维护。

优点：

高度可维护和可测试
松散耦合
可独立部署
围绕业务能力进行组织

缺点：

服务数量大幅增加，部署维护困难
服务间依赖管理
服务故障处理

容器

那么我们提到了项目演进，在同一时间，容器技术的标准化统一也间接促成了微服务的推广（我猜的），Docker 在 2013.03.13 发布第一个版本，容器化技术让我们产品发布形态有了新的选择，开发直接将容器镜像发布，运维同学通过镜像进行产品上线，确保了环境的统一，无须纠结环境配置相关问题（不用吵架了）。

当我们产品发布采用容器化上线后，我们面临一些其他的问题了：

微服务追求的是将服务解耦，拆分为多个服务，那么最终发布形态对应的也是多个镜像，运维同学管理这些镜像之间的关系难度增加。
同时当镜像运行在不同的物理节点上，对计算资源和网络资源的要求是一致的，运维同学需要做到让镜像无感知。
当产品要进行升级时，镜像之间的依赖关系，故障切换等操作紧靠现有容器功能实现困难。

这么一看，与之前单体应用比也没好哪里去。于是有了各种容器编排系统，比如 Swarm，Mesos等等，但都不是很好用且各家一个标准，这时候老大哥谷歌发话了，我来把我们内部用了很多年要淘汰的东西拿出来给大家解决问题吧，于是有了 Kubernetes。

Kubernetes 功能上提供了解决微服务引入的问题，并更好的配合微服务去提供稳定高可用的统一容器化环境，具体如何解决的我们后续可以通过了解 Pod，ConfigMap，ReplicaSet 等功能去详细了解。

总结

可能是因为我考虑问题都是从运维角度去看的，网上的一些文章讲的带来的好处反而没看太清，可能作为 2C 产品，追求敏捷开发，产品不断快速迭代的目标比较适合，但是如果本身作为一个追求稳定可靠的 2B 产品来说，引入 k8s 带来的好处和维护 k8s 带来的成本真的要仔细的从产品层面考虑清楚，这里感觉跟具体的技术关系不大，而是说从产品面向的客户对象考虑，客户想要的是一个什么产品，而 k8s 作为一个还在不断（频繁）迭代的产品来说（可以去看看 release notes 的更新速度），后续若出现某些 API 不兼容等情况，如何去应对，感觉还是个灾难。

参考链接

在学习过程中，通过阅读 cch123 的博客对微服务有了更深的了解，这里列一下相关的系列博客链接：

Kubernetes 实战-前言

发表于 2019-05-11 |

Kubernetes 实战-前言

自从 Kubernetes 大热之后，一直没跟着版本去了解具体的功能及使用，只是大概了解其中概念。之前推特上有人推荐《Kubernetes In Action》这本书，说是对入门同学很友好，利用五一假期和这个周末，终于看完了，打算把学习过程和其中的一些想法记录下来。

《Kubernetes In Action》

就想推荐人说的那样，这本书作为 101 系列来说，是很称职的，你跟着官方示例做，90%以上都是可以成功的，且讲解门槛不高，推荐。

中文版是由七牛团队翻译的，虽然其中有一些小的翻译错误，但是整体读下来还是很顺畅的，不影响阅读，当然现在网上已经有原版资源，想读的同学可以去 SaltTiger 搜索下载。

本书章节较多，分为 3 部分：What？How？Why？首先讲解 k8s 及容器的基本概念，然后讲解 k8s 基本使用，最后介绍了一些 k8s 工作原理及最佳实践。

当你了解了什么是 k8s，及 k8s 能带来什么好处之后，我们去使用 k8s，从而真实的感受到 k8s 带来的便利，这种感觉是很美好的（表面美好的东西肯定会有某些限制），对我来说这种美好截止到第二部分就停止了。在第三部分中，我们在之前感受到的便利隐藏着很多没有考虑到的边界因素，也意味着我们从一个传统的单节点服务切换到微服务架构上，会新增很多需要去考虑的因素，如果 k8s 内部提供了解决方案，那么很简单，我们直接编写 YAML 就可以了，如果 k8s 没有解决方案呢？我不知道，可能当我真正体验过之后才能给出感受吧（即将发生的事）。

下一篇我们来说下什么是微服务。

LeetCode Shell 题解

发表于 2019-05-11 |

工作上用 Shell 的频率是很高的，哪怕现在有了 Ansible 或者其他配置工具，Shell 仍是一个以 Linux 作为工作环境的同学的必备技能。
之前写过 GitHub 上的 Pure Bash Bible 的博客，看到 LeetCode 上的 Shell 题目好久不更新了，只有 4 道，今天记录一下题解。

192. Word Frequency

统计文本文件中单词出现次数，倒序输出。

words.txt

1 2	the day is sunny the the the sunny is is

利用 tr sort uniq awk 解决。

1 2	# Read from the file words.txt and output the word frequency list to stdout. cat words.txt \| tr -s ' ' '\n' \| sort \| uniq -c \| sort -rn \| awk '{ print $2, $1 }'

193. Valid Phone Numbers

校验电话号码正确性。

file.txt

1
2
3

987-123-4567
123 456 7890
(123) 456-7890

主要是利用 grep 匹配正则，注意转义符。

1 2	# Read from the file file.txt and output all valid phone numbers to stdout. grep -P '^(\d{3}-\|$\d{3}$ )\d{3}-\d{4}$' file.txt

194. Transpose File

行和列转换。

file.txt

1
2
3

name age
alice 21
ryan 30

# Read from the file file.txt and print its transposed content to stdout.
ncol=`head -n1 file.txt | wc -w`

for i in `seq 1 $ncol`
do
    echo `cut -d' ' -f$i file.txt`
done

195. Tenth Line

显示文件第 10 行。

file.txt

Line 1
Line 2
Line 3
Line 4
Line 5
Line 6
Line 7
Line 8
Line 9
Line 10

如果直接使用 head tail 的话不能方便处理文件为空的情况。

1 2	# Read from the file file.txt and output the tenth line to stdout. sed -n '10p' file.txt

总结

其实编写 Shell 在熟知 Linux 内置命令就可以处理大部分场景了，如果处理不来，只能求助于 sed 和 awk, 但我脑子可能不太好使，awk 的语法总是记不住，每次写之前都要查一下语法。 - -

如果不想在代码中充斥着各种转义处理的话，还是老老实实使用 Python 编写脚本吧。

PySnooper 源码阅读

发表于 2019-04-27 |

背景

在 18 年的时候 jiajun 同学发过一篇博客，讲如何调试相关的总结。结合最近自己的经验，紧靠 logging 和 print 就能解决日常的 80%问题，剩下的 20% 也都可以通过review 代码来解决，我只有当确实没什么思路的时候，才会采用 pdb 的方式去调试。之所以先 review 代码再采用 pdb 的方式是想确认自己已经理清了相关代码的上下文和逻辑，不至于在单步调试的时候出现 恍然大悟 （贬义）的状况。

最近两天 Github 上关于 Python 的项目最火的就是 PySnooper，这个项目的 Slogan 就是 Never use print for debugging again ，这里的 print 替换为 logging 也没啥差。整个代码在初步可用阶段代码量很少，也确实能够给平时写些小脚本带来便利，便抽时间看了看具体的实现。

PySnooper

先来看下目录结构：

yiran@zhouyirandeMacBook-Pro:~/Documents/git-repo/PySnooper
3d0d051 ✗ $ tree .
.
├── LICENSE
├── MANIFEST.in
├── README.md
├── make_release.sh
├── misc
│   └── IDE\ files
│       └── PySnooper.wpr
├── pysnooper
│   ├── __init__.py
│   ├── pycompat.py
│   ├── pysnooper.py
│   ├── tracer.py
│   └── utils.py
├── requirements.in
├── requirements.txt
├── setup.py
├── test_requirements.txt
└── tests
    ├── __init__.py
    ├── test_pysnooper.py
    └── utils.py

可以看到最核心部分都在 pysnooper 部分，我们以官方示例来了解具体是如何工作的:

root@yiran30250:~/backup/PySnooper
master ✗ $ cat a.py
import pysnooper

@pysnooper.snoop('/var/log/test.log')
def number_to_bits(number):
    print('func starting...')
    if number:
        bits = []
        while number:
            number, remainder = divmod(number, 2)
            bits.insert(0, remainder)
        return bits
    else:
        return [0]

print(number_to_bits(6))

从示例中可以看到， pysnooper.snoop 作为一个装饰器，装饰所需要调试的函数，并可以再装饰器参数中添加对应的输出目的，比如标准输出，或者指定日志等。

我们看下 snoop 的实现：

def snoop(output=None, variables=(), depth=1, prefix='', overwrite=False):
    write, truncate = get_write_and_truncate_functions(output) # 通过输出目标获取 write 函数
    if truncate is None and overwrite:
        raise Exception("`overwrite=True` can only be used when writing "
                        "content to file.")
    def decorate(function):
        target_code_object = function.__code__ # 函数在 python 解释器编译后的字节码对象
        tracer = Tracer(target_code_object=target_code_object, write=write,
                        truncate=truncate, variables=variables, depth=depth,
                        prefix=prefix, overwrite=overwrite)
        # 实例化 Tracer，将现有参数全部传递
        def inner(function_, *args, **kwargs):
            with tracer: # 通过 with 关键字调用 tracer，那么 Tracer 内应该实现了上下文管理器的 `__enter__` 和 `__exit__` 方法
                return function(*args, **kwargs) #
        return decorator.decorate(function, inner)

    return decorate

主要功能应该在 Trancer 中实现的，我们看下 Trancer 中做了什么？

def __enter__(self):
    self.original_trace_function = sys.gettrace()
    sys.settrace(self.trace)

def __exit__(self, exc_type, exc_value, exc_traceback):
    sys.settrace(self.original_trace_function)

先忽略其他的，我们先看实现上下文管理器的方法:

进入上下文环境
- 获取当前跟踪器并记录
- 设置追踪器为 self.trace
退出上下文环境
- 将追踪器设置为原有值

注意：

sys.settrace 官方文档中描述它只用来做调试类工具，不建议在内部实现复杂逻辑。
The gettrace() function is intended only for implementing debuggers, profilers, coverage tools and the like.

其中追踪器要接受 3 个参数，分别是：frame，event 和 arg。我们先记住 frame就是当前的栈帧就好。

看一下 self.trace 具体是如何工作的，先看第一部分：

def trace(self, frame, event, arg):
    # 这里的注释写的很清楚了，根据当前 frame 是否为指定的函数字节码对象，如果不是且深度为 1，则直接返回 trace，如果指定了追踪深度，则不断循环，直到追踪到指定函数字节码对象
    if frame.f_code is not self.target_code_object:
        if self.depth == 1:
            return self.trace
        else:
            _frame_candidate = frame
            for i in range(1, self.depth):
                _frame_candidate = _frame_candidate.f_back # f_back 为当前栈帧的上一个栈帧，便于在当前代码执行完成后可以调回之前代码继续执行
                if _frame_candidate is None:
                    return self.trace
                elif _frame_candidate.f_code is self.target_code_object:
                    indent = ' ' * 4 * i
                    break
            else:
                return self.trace
    else:
        indent = ''

找到了具体的执行对象，我们看下如何获取环境变量的：

def trace(self, frame, event, arg):
    ...
    self.frame_to_old_local_reprs[frame] = old_local_reprs = \
                                           self.frame_to_local_reprs[frame] # 标记当前变量为现有变量
    self.frame_to_local_reprs[frame] = local_reprs = \
                           get_local_reprs(frame, variables=self.variables) # 获取当前变量

    modified_local_reprs = {}
    newish_local_reprs = {}

    for key, value in local_reprs.items(): # 遍历当前本地变量，将其分别放置为新变量和修改变量列表中
        if key not in old_local_reprs:
            newish_local_reprs[key] = value
        elif old_local_reprs[key] != value:
            modified_local_reprs[key] = value
    # 将变量通过 write 函数输出到对应的目标中
    newish_string = ('Starting var:.. ' if event == 'call' else
                                                        'New var:....... ')
    for name, value_repr in sorted(newish_local_reprs.items()):
        self.write('{indent}{newish_string}{name} = {value_repr}'.format(
                                                               **locals()))
    for name, value_repr in sorted(modified_local_reprs.items()):
        self.write('{indent}Modified var:.. {name} = {value_repr}'.format(
                                                               **locals()))

在上面可以看到大部分都是判断逻辑，看一下 get_local_reprs 中是如何获取当前变量的：

def get_local_reprs(frame, variables=()):
    result = {key: get_shortish_repr(value) for key, value
                                                     in frame.f_locals.items()}
    for variable in variables:
        try:
            result[variable] = get_shortish_repr(
                eval(variable, frame.f_globals, frame.f_locals)
            )
        except Exception:
            pass
    return result

是通过调用 frame.f_locals.items() 获取当前栈帧所具有的本地变量。
剩下的部分是对于调试函数为装饰器时，需要特殊处理：跳过装饰器函数，直到找到函数定义部分。

总结

主题功能就通过上述代码来实现，简单高效，我们再来回顾一下：

通过 settrace 来设置追踪器
a. settrace 的行为是先执行追踪器部分，执行完成后执行函数字节码对应行
在追踪器中，通过 frame 相关属性来获取所需值，如 f_locals, f_code, f_globals
打印相关信息到目标中，退出上下文管理器，重新设置外层追踪器

PingCAP tidb-ansible 源码阅读

发表于 2019-04-20 |

了解我的同学应该知道，我目前负责公司产品的运维工具开发相关的工作，作为一款 2B 的产品，在产品运维过程中，总是有一些不愉快（又不能让客户知道）的繁琐操作：这些操作可能是为了防止过程中出现错误，而不断添加的检测条件；也有可能是历史问题，随着产品发布迭代而一直遗留至今。

所以我平时也在关注一些开源的 2B 产品的配套运维工具，比如 ZStack、PingCAP 之类的公司。

但是 ZStack 的开源生态不是很好，感觉只是在保持代码更新（不知道哪个分支）的状态。相比之下 PingCAP 就好很多了，可以很直接的从文档中看到差别，而且社区很活跃。

最近看到 PingCAP 的一个关于部署维护的视频讲解，可能时间有限，并没有很深入的讲解细节，有兴趣的同学可以看下。

在讲解过程中，一个比较核心的工具就是 Ansible，通过 Ansible Playbook 来定义各个步骤，我最近也在使用 Ansible 来进行二次开发，特此学习下 PingCAP 的 tidb-ansible 。

安装方式

TiDB 目前支持 4 种安装方式：

Ansible Online
Ansible Offline
Docker
Docker compose

其中最佳实践应该是 Ansible Online 方式，通过控制机联网下载所需依赖及 TiDB binary 文件。当然如果所在环境无法访问互联网，那么只能采用 Offline 方式了。后两种部署方式，感觉只是用于开发测试或者给用户“看看”的情况。

如果要学习的话肯定要学习最佳实践了，那么我们来看看 Ansible Online 方式。

环境准备

TiDB 作为一个 开源分布式关系型数据库 ，所需要的物理环境是很比较苛刻的，官方最佳实践的需求如下：

组件	CPU	内存	硬盘类型	网络	数量(最低要求)
TiDB	16	32GB	SAS	10GbE * 2	2
PD	4	8GB	SSD	10 GbE * 2	3
TiKV	16	32GB	SSD	10 GbE * 2	3
监控	8	16GB	SAS	1GbE * 1	1

对于 CPU、内存和磁盘的要求我们暂时忽略，这里注意网卡数量都是推荐的 2 块网卡，应该是会做 bonding，到时候看下代码中是否处理。

安装部署流程

控制节点安装依赖
配置普通用户（tidb）
- 密码
- sudu 权限
- ssh key
- …
配置控制节点到被安装节点 SSH 免密登录
- tidb 免密
NTP
- 各节点时钟同步
CPU 调节器模式
- performance
格式化磁盘
- 推荐 ext4，支持 xfs
- 挂载参数：nodelalloc 和 noatime
ansible inventory & TiDB 配置修改
- inventory
- tidb-ansible/conf/*
部署
- local_prepare.yml
- bootstrap.yml
- deploy.yml
- start.yml

具体步骤

软件依赖

这里 TiDB 依赖的第三方组件不多，安装方式猜测之所以直接用 pip 是因为要支持不同的发型版本，python 通用些。
其中如果部署 pump 并开启 binlog 的话，是需要安装 Kafka 集群的，这里在安装部署文档中没有见到更多的说明。

普通用户

这点很重要，一款软件的运行环境很重要，无须 root 权限的坚决不能给与 root 权限，否则之后的权限控制很难做。

时钟同步

这里采用的是 ntpd 同步，不知道为什么没有采用 chronyd，在 RHEL7 之后推荐采用 chronyd 来进行时钟同步，根据这篇博客中提到的，chronyd 应该占有绝对优势的，不知道这里是出于什么考虑。

CPU 调节器模式

这里 TiDB 推荐采用 performance 模式，但是如果你的 2B 产品卖点中有提到 节省能耗 相关字眼的，觉得还是要综合考虑才好。

格式化磁盘

推荐采用 ext4，在 RHEL7 之后默认的系统分区均为 xfs，这里出于什么考虑也没有透露。

在我个人使用过程中，xfs 在处理服务器掉电之后的处理很麻烦，经常是需要手动 xfs_repair 去修复磁盘分区。

TiDB 挂载参数强制要求 nodelalloc ，我们看下这个参数的意义：

delayed allocation. Blocks are allocated

1	when the data is copied from userspace to the page cache, either via the write(2) system call or when an mmap'ed page which was previously unallocated is written for the first time.

TiDB 参数

因为对 TiDB 无更多了解，此节忽略。

部署

终于到了我最想了解的地方了：ansible playbook。我们先看下代码结构：

yiran@zhouyirandeMacBook-Pro:~/Documents/git-repo/tidb-ansible
master ✔ $ tree -L 2 .
.
├── LICENSE
├── README.md
├── ansible.cfg
├── bootstrap.yml
├── clean_log_cron.yml
├── cloud
│   └── aws-ansible
├── collect_diagnosis.yml
├── common_tasks
│   ├── add_evict_leader_scheduler.yml
|   ...
│   └── transfer_pd_leader.yml
├── conf
│   ├── alertmanager.yml
|   ...
│   └── tikv.yml
├── create_users.yml
├── deploy.yml
├── deploy_drainer.yml
├── deploy_ntp.yml
├── excessive_rolling_update.yml
├── filter_plugins
│   └── tags.py
├── graceful_stop.yml
├── group_vars
│   ├── alertmanager_servers.yml
│   ├── all.yml
|   ...
│   ├── tidb_servers.yml
│   └── tikv_servers.yml
├── hosts.ini
├── inventory.ini
├── library
│   ├── coreos_facts
│   ├── docker_facts
│   └── wait_for_pid.py
├── local_prepare.yml
├── log
├── migrate_monitor.yml
├── requirements.txt
├── roles
│   ├── alertmanager
│   ├── blackbox_exporter
│   ├── bootstrap
|   ...
│   ├── tikv
│   ├── tikv_importer
│   └── tispark
├── rolling_update.yml
├── rolling_update_monitor.yml
├── scripts
│   ...
│   ├── check
│   ├── clsrun.sh
│   ├── disk_performance.json
│   ├── grafana-config-copy.py
|   ...
│   └── tikv_trouble_shooting.json
├── start.yml
├── start_drainer.yml
├── start_spark.yml
├── stop.yml
├── stop_drainer.yml
├── stop_spark.yml
├── templates
│   └── grafana.dest.json.j2
├── unsafe_cleanup.yml
├── unsafe_cleanup_container.yml
└── unsafe_cleanup_data.yml

一个标准的 Ansible Playbook 结构：在最外层暴露我们需要执行的 YAML 配置，所有具体的操作和配置文件都放到 roles 和 conf 中，下面我们来一点点看具体做了什么，哪些地方是我们需要注意的。

local_prepare.yml

在这里一共做了以下这么几件事情：

准备 binary 下载地址配置
检查网络(GFW)
下载 binary
准备 fio 软件
清理下载路径

主要都是做环境准备的工作，其中网络检测这没有用我们日常使用最多的 ping ，而是使用了 curl ，这里猜测是因为有时候机器是可以与互联网通信，但是没有配置 DNS 解析，导致后续的下载也会失败，所以直接 curl baidu.com 是一个不错的选择。

(这里还检查了 GFW - - )

用到的 Ansible 模块有：

file
shell （果然没人会去用 command 的
template
get_url
…

bootstrap.yml

基本环境配置参数检查（机器数、配置、OS发行版）
python 环境准备，NTP 软件包安装
设置内核参数
检查系统配置（cpu 调节器配置如果不是performance，非开发模式下会强制退出）
fio 检测
- psync ,bs=32k,iodepth=4,numjobs=4 randread iops不小于 40k
- psync, bs=32k,iodepth=4,numjobs=4 randrw(100%r,0%w) iops均不低于 10k
- psync, bs=32k, iodepth=1,numjobs=1 randrw(100%r,0%w) lat randread 不高于 0.25ms，write 不高于 30us

内核参数部分修改了如下配置：

net.core.somaxconn=32768 # 最大 socket 连接数，默认为 128
vm.swappiness=0 # 禁用 swap 空间
net.ipv4.tcp_syncookies=0 # 关闭 synccookies，默认为关闭
net.ipv4.tcp_tw_recycle=0 # TIME_WAIT 快速回收，默认关闭，关于 TIME_WAIT 状态可以在《TCP/IP 详解卷一协议》中了解更多
fs.file-max=1000000 # 最大文件数，默认为 1024
irqbalance ONESHOT=yes # irqbalance 不以守护进程方式运行

关于 fio 检测，嗯，我还是太天真了，这个级别的磁盘延迟要求，估计只有 PCIe SSD 可以满足了吧。

deploy.yml

这部分通用的东西比较少，都是内部的一些配置文件和服务。

start.yml

貌似 TiDB 服务的启动是有严格的依赖关系的，不知道这里如果启动顺序调整，是否会自动重试直到相应依赖服务启动完成。

滚动升级

在视频讲解中，提到 TiDB 支持滚动升级，这一点在一个 2B 产品功能中很重要，随着产品的开发迭代，能够无痛的给客户进行升级是很重要的。

在官方文档中，看到了两个升级文档，分别为 1.0 升级到 2.0 和 2.0 升级到 2.1。使用的滚动升级 YAML 文件为：rolling_update.yml
升级过程为滚动升级，步骤为：

pre check
upgrade
post upgrade

这里没搞懂为什么 pre check tidb-server 也要顺序执行，看上去只是检测了 TiDB 版本。

在 YAML 中指定的升级顺序为：

pd
tikv
pump
tidb

这里比较遗憾没有找到升级失败的处理方式，升级 PD 过程中检查健康状态连续 1分钟都未恢复正常，此时 Ansible 退出，PD 处于高版本状态，其他组件 tikv，tidb 还处于低版本状态，不知道这时如何处理，也许是产品上保证了兼容性？

总结

了解了 TiDB 的安装和升级，着重学习下 Ansible Playbook 的组织形式，在 tidb-ansible 中，所有的功能都拆分的很细，能采用 ansible 的都用 ansible 实现了，但是难免需要一些脚本配合（fio），这里的选择需要根据实际情况来进行相应修改。

吐槽

不知道为什么在代码中 scripts 下放置了许多json 配置文件。。。。

记一次 libcgroup 配置失败

发表于 2019-04-11 |

cgroup 配置失败解决方案

看过之前博客的同学应该知道，我一直使用的 libcgroup 来进行 cgroup 配置，简单方便。
最近遇到了一个报错，很坑，记录一下。

报错

接到反馈说有个环境在产品升级之后， cgconfig.service 无法启动，当时的配置如下：

[root@yiran-test 21:31:59 ~]$cat /etc/cgconfig.conf
# yiran cgroups configuration

group . {
    cpuset {
        cpuset.memory_pressure_enabled = "1";
    }
}

group yiran {
    cpuset {
        cpuset.cpus = "0,1,2,3,4,5";
        cpuset.mems = "0-1";
        cpuset.cpu_exclusive = "1";
        cpuset.mem_hardwall = "1";
    }
}

group yiran/bb-main {
    cpuset {
        cpuset.cpus = "0";
        cpuset.mems = "0-1";
        cpuset.cpu_exclusive = "1";
        cpuset.mem_hardwall = "1";
    }
}

group yiran/bb-io {
    cpuset {
        cpuset.cpus = "1";
        cpuset.mems = "0-1";
        cpuset.cpu_exclusive = "1";
        cpuset.mem_hardwall = "1";
    }
}

group yiran/aa-main {
    cpuset {
        cpuset.cpus = "2";
        cpuset.mems = "0-1";
        cpuset.cpu_exclusive = "1";
        cpuset.mem_hardwall = "1";
    }
}

group yiran/others {
    cpuset {
        cpuset.cpus = "3";
        cpuset.mems = "0-1";
        cpuset.cpu_exclusive = "1";
        cpuset.mem_hardwall = "1";
    }
}


group yiran/app {
    cpuset {
        cpuset.cpus = "4,5";
        cpuset.mems = "0-1";
        cpuset.cpu_exclusive = "0";
        cpuset.mem_hardwall = "1";
    }
}

group qemu {
    cpuset {
        cpuset.cpus = "6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25,26,27,28,29,30,31,32,33,34,35,36,37,38,39,40,41,42,43,44,45,46,47";
        cpuset.mems = "0-1";
        cpuset.cpu_exclusive = "0";
        cpuset.mem_hardwall = "1";
    }
}

咋一看，配置文件看上去是正确的，除了最后一个组的 cpuset.cpus 配置略长，但是也没错，按道理应该服务正常启动才对，尝试重启服务查看服务报错信息：

[root@yiran-test 21:33:55 ~]$systemctl restart cgred
[root@yiran-test 21:33:59 ~]$systemctl restart cgconfig
Job for cgconfig.service failed because the control process exited with error code. See "systemctl status cgconfig.service" and "journalctl -xe" for details.
[root@yiran-test 21:34:05 ~]$systemctl status cgconfig
● cgconfig.service - Control Group configuration service
   Loaded: loaded (/usr/lib/systemd/system/cgconfig.service; enabled; vendor preset: disabled)
   Active: failed (Result: exit-code) since 四 2019-04-11 21:34:05 CST; 4s ago
  Process: 6744 ExecStop=/usr/sbin/cgclear -l /etc/cgconfig.conf -L /etc/cgconfig.d -e (code=exited, status=3)
  Process: 11465 ExecStart=/usr/sbin/cgconfigparser -l /etc/cgconfig.conf -L /etc/cgconfig.d -s 1664 (code=exited, status=109)
 Main PID: 11465 (code=exited, status=109)

4月 11 21:34:05 yiran-test systemd[1]: Starting Control Group configuration service...
4月 11 21:34:05 yiran-test cgconfigparser[11465]: /usr/sbin/cgconfigparser; error loading /etc/cgconfig.conf: Failed to remove a non-empty group
4月 11 21:34:05 yiran-test cgconfigparser[11465]: Error: failed to set /sys/fs/cgroup/cpuset/qemu/cpuset.cpus: Invalid argument
4月 11 21:34:05 yiran-test systemd[1]: cgconfig.service: main process exited, code=exited, status=109/n/a
4月 11 21:34:05 yiran-test systemd[1]: Failed to start Control Group configuration service.
4月 11 21:34:05 yiran-test systemd[1]: Unit cgconfig.service entered failed state.
4月 11 21:34:05 yiran-test systemd[1]: cgconfig.service failed.

发现在重启 cgconfig 时报错，报错信息如下：

1
2

4月 11 21:34:05 yiran-test cgconfigparser[11465]: /usr/sbin/cgconfigparser; error loading /etc/cgconfig.conf: Failed to remove a non-empty group
4月 11 21:34:05 yiran-test cgconfigparser[11465]: Error: failed to set /sys/fs/cgroup/cpuset/qemu/cpuset.cpus: Invalid argument

第一行说移除一个非空的 cgroup 失败，下一条提示 /sys/fs/cgroup/cpuset/qemu/cpuset.cpus 也就是我们觉得略微异常的 cgroup 参数无效，可是参数明明是正确配置的，为啥就无效了呢？

调查

尝试修改 cpuset.cpus ，将其调整为：

[root@yiran-test 21:36:32 ~]$tail /etc/cgconfig.conf

group qemu {
    cpuset {
        cpuset.cpus = "6,7,8,9,10";
        cpuset.mems = "0-1";
        cpuset.cpu_exclusive = "0";
        cpuset.mem_hardwall = "1";
    }
}

重启 cgconfig 服务：

[root@yiran-test 21:36:36 ~]$systemctl restart cgred
[root@yiran-test 21:36:55 ~]$systemctl restart cgconfig
[root@yiran-test 21:37:00 ~]$systemctl status cgconfig
● cgconfig.service - Control Group configuration service
   Loaded: loaded (/usr/lib/systemd/system/cgconfig.service; enabled; vendor preset: disabled)
   Active: active (exited) since 四 2019-04-11 21:37:00 CST; 3s ago
  Process: 6744 ExecStop=/usr/sbin/cgclear -l /etc/cgconfig.conf -L /etc/cgconfig.d -e (code=exited, status=3)
  Process: 17491 ExecStart=/usr/sbin/cgconfigparser -l /etc/cgconfig.conf -L /etc/cgconfig.d -s 1664 (code=exited, status=0/SUCCESS)
 Main PID: 17491 (code=exited, status=0/SUCCESS)

4月 11 21:37:00 yiran-test systemd[1]: Starting Control Group configuration service...
4月 11 21:37:00 yiran-test systemd[1]: Started Control Group configuration service.

服务正常运行了，那么我推测可能是跟 cpuset.cpus 长度有关，这时候只能求助于 Google 啦。

很容易，我们找到了这个答案，RedHat 官方 KB 中的介绍：

Root Cause
Previously, the internal representation of a value of any cgroup subsystem parameter was limited to have the length of 100 characters at maximum. Consequently, the libcgroup library truncated the values longer than 100 characters before writing them to a file representing matching cgroup subsystem parameter in the kernel.

Resolution
The maximal length of values of cgroup subsystem parameters in libcgroup has been extended to 4096 characters. As a result, libcgroup now handles values of cgroup subsystem parameters with any length correctly. (BZ#1549175)
RHEL 6 https://access.redhat.com/downloads/content/rhel---6/x86_64/169/libcgroup/0.40.rc1-26.el6/src/fd431d51/package (or newer) via RHBA-2018:1861
RHEL 7 libcgroup-0.41-20.el7 (or newer) via RHBA-2018:3058

官方给出的解决方式是通过升级 libcgroup 来解决，但是我不想这么做。

为什么？要知道在生产环境中，我们想要进行第三方软件包的升级是要经过层层测试的，等到测试完成不知道什么时候了，所以我们需要一个快速折中方案。

解决方案

我们知道，libcgroup 只是作为一个配置 cgroup 软件的一种，最终操作的都是 cgroup 实际挂载点下的配置文件，比如 CentOS 默认的 /sys/fs/cgroup/ 。

那么我们来看下正常配置下 qemu 组中的 cpuset.cpus 配置长什么样：

1 2	[root@yiran-test 21:37:04 ~]$cat /sys/fs/cgroup/cpuset/qemu/cpuset.cpus 6-10

Ok，我们明明在 libcgroup 配置文件中写的是 6,7,8,9,10 ,在 cgroup 配置文件中就转换成了 6-10，那么一切都简单了。

当我们配置 cgroup 如果 cpu processor id 是连续的，那么我们就可以通过 - 来连接起始和终止 id 就可以了，问题解决。

总结

libcgroup 作为 RedHat 官方指定的 cgroup 配置工具，没想到会出现这种问题，如果 cpuset.cpus 我们要设置非连续的 cpu processor id 的话，只能通过升级方式解决了。

apscheduler 源码阅读

发表于 2019-04-06 |

简介

apscheduler 全称 Advanced Python Scheduler，调度器，主要功能如下：

动态添加、删除任务
暂停、恢复任务
周期性调度：cron,date,interval
…

那么接下来我们根据官方示例，看看 apscheduler 是如何进行处理任务的。

示例版本为 2.1，因为在 2.1 版本包含目前 master 分支上的主要功能，简单易懂。

代码结构如下：

yiran@zhouyirandeMacBook-Pro:~/Documents/git-repo/apscheduler
2.1 ✔ $ tree apscheduler
apscheduler
├── __init__.py
├── events.py
├── job.py
├── jobstores
│   ├── __init__.py
│   ├── base.py
│   ├── mongodb_store.py
│   ├── ram_store.py
│   ├── redis_store.py
│   ├── shelve_store.py
│   └── sqlalchemy_store.py
├── scheduler.py
├── threadpool.py
├── triggers
│   ├── __init__.py
│   ├── cron
│   │   ├── __init__.py
│   │   ├── expressions.py
│   │   └── fields.py
│   ├── interval.py
│   └── simple.py
└── util.py

示例：

from datetime import datetime, timedelta

from apscheduler.scheduler import Scheduler
from apscheduler.jobstores.shelve_store import ShelveJobStore


def alarm(time):
    print('Alarm! This alarm was scheduled at %s.' % time)


if __name__ == '__main__':
    scheduler = Scheduler(standalone=True)
    scheduler.add_jobstore(ShelveJobStore('example.db'), 'shelve')
    alarm_time = datetime.now() + timedelta(seconds=10)
    scheduler.add_date_job(alarm, alarm_time, name='alarm',
                           jobstore='shelve', args=[datetime.now()])
    print('To clear the alarms, delete the example.db file.')
    print('Press Ctrl+C to exit')
    try:
        scheduler.start()
    except (KeyboardInterrupt, SystemExit):
        pass

Scheduler

上述示例很容易理解，首先对 Scheduler 实例化，然后添加 jobstore，定义一个名为 alarm 的 job，并指定其运行时间为当前时间 + 10s，将该 job 添加到 scheduler 中，添加 job 类型为 date_job，然后启动 scheduler。

可以看到我们所有的操作都是通过 scheduler 方法实现的，那么我们来看下 Scheduler 类具体实现了哪些功能：

class Scheduler(object):
    """
    This class is responsible for scheduling jobs and triggering
    their execution.
    """

    _stopped = True
    _thread = None

    def __init__(self, gconfig={}, **options):
        self._wakeup = Event()
        self._jobstores = {}
        self._jobstores_lock = Lock()
        self._listeners = []
        self._listeners_lock = Lock()
        self._pending_jobs = []
        self.configure(gconfig, **options)

在 Scheduler 构造函数中，对一些变量进行初始化，这里要注意 self._wakeup ，后续的一些主要功能都是通过它来实现的。接下来看看 add_jobstore 方法：

def add_jobstore(self, jobstore, alias, quiet=False):
    self._jobstores_lock.acquire() # 请求锁
    try:
        if alias in self._jobstores:
            raise KeyError('Alias "%s" is already in use' % alias)
        self._jobstores[alias] = jobstore # 将 jobstore 别名作为 key，添加到 self._jobstores 中
        jobstore.load_jobs() # 加载 jobstore 中所有 job
    finally:
        self._jobstores_lock.release() # 释放锁

    # Notify listeners that a new job store has been added
    self._notify_listeners(JobStoreEvent(EVENT_JOBSTORE_ADDED, alias)) # 事件通知

    # Notify the scheduler so it can scan the new job store for jobs
    if not quiet:
        self._wakeup.set() # 将 Event 置为 True

在 add_jobstore 中，将 jobstore 添加到 scheduler 中，并加载当前 jobstore 中的所有任务，接下来将具体的 job 添加到 scheduler 中：

1
2
3

def add_date_job(self, func, date, args=None, kwargs=None, **options):
    trigger = SimpleTrigger(date)
    return self.add_job(trigger, func, args, kwargs, **options)

这里的 SimpleTrigger 只是多种 Trigger 中的一种，根据 Trigger 类型的不同，最主要的差别在于 get_next_fire_time 计算方式不同。

如果添加的任务是 interval_job，那么对应 Trigger 为 IntervalTrigger ；如果添加的任务是 cron_job，那么对应的 Trigger 为 CronTrigger。

可以看到不同的任务只是 Trigger 计算方式不同，最终还是通过 add_job 方法，继续看：

def add_job(self, trigger, func, args, kwargs, jobstore='default',
            **options):
    job = Job(trigger, func, args or [], kwargs or {},
              options.pop('misfire_grace_time', self.misfire_grace_time),
              options.pop('coalesce', self.coalesce), **options) # 将 job 实例化
    if not self.running: # 如果 scheduler 未启动，那么将其添加到等待队列中
        self._pending_jobs.append((job, jobstore))
        logger.info('Adding job tentatively -- it will be properly '
                    'scheduled when the scheduler starts')
    else:
        self._real_add_job(job, jobstore, True) # 否则添加 job 到 jobstore 中
    return job

继续看 _real_add_job 中的实现：

def _real_add_job(self, job, jobstore, wakeup):
    job.compute_next_run_time(datetime.now()) # 计算job 下次运行时间
    if not job.next_run_time:
        raise ValueError('Not adding job since it would never be run')

    self._jobstores_lock.acquire() # 请求锁
    try:
        try:
            store = self._jobstores[jobstore]
        except KeyError:
            raise KeyError('No such job store: %s' % jobstore)
        store.add_job(job) # 添加 job
    finally:
        self._jobstores_lock.release()

    # Notify listeners that a new job has been added
    event = JobStoreEvent(EVENT_JOBSTORE_JOB_ADDED, jobstore, job)
    self._notify_listeners(event) # 事件通知

    logger.info('Added job "%s" to job store "%s"', job, jobstore)

    # Notify the scheduler about the new job
    if wakeup:
        self._wakeup.set() # # 将 Event 置为 True

在 _real_add_job 中我们终于看到 store.add_job(job) ，至于 store.add_job 如何实现我们之后看 JobStore 再说。

现在我们已经给 scheduler 添加了 jobstore 和 job，那么看下 scheduler 是如何运行的：

def start(self):
    ...
    # Schedule all pending jobs
    for job, jobstore in self._pending_jobs: # 将 scheduler 未运行时添加的 job，即在等待队列中的 job 添加到 jobstore 中
        self._real_add_job(job, jobstore, False)
    del self._pending_jobs[:]

    self._stopped = False
    if self.standalone:
        self._main_loop()
    else:
        self._thread = Thread(target=self._main_loop, name='APScheduler')
        self._thread.setDaemon(self.daemonic)
        self._thread.start()

在 scheduler 运行时，会先将所有的 job 加载到 jobstore 中，然后调用 self._main_loop ，如果 Standalone 为 True，则会一直阻塞知道没有 job 需要运行，看看 self._main_loop 做了啥：

def _main_loop(self):
    """Executes jobs on schedule."""

    logger.info('Scheduler started')
    self._notify_listeners(SchedulerEvent(EVENT_SCHEDULER_START)) # 事件通知 

    self._wakeup.clear()
    while not self._stopped:
        logger.debug('Looking for jobs to run')
        now = datetime.now()
        next_wakeup_time = self._process_jobs(now) # 计算下次唤醒时间

        if next_wakeup_time is not None:
            wait_seconds = time_difference(next_wakeup_time, now)
            logger.debug('Next wakeup is due at %s (in %f seconds)',
                         next_wakeup_time, wait_seconds)
            try:
                self._wakeup.wait(wait_seconds) # 等待 Event flag
            except IOError:  # Catch errno 514 on some Linux kernels
                pass
            self._wakeup.clear()
        elif self.standalone:
            logger.debug('No jobs left; shutting down scheduler')
            self.shutdown() # 若 scheduler standalone 为 True 且 jobs 为空，则停止 scheduler
            break
        else:
            logger.debug('No jobs; waiting until a job is added')
            try:
                self._wakeup.wait() # 等待 Event flag
            except IOError:  # Catch errno 514 on some Linux kernels
                pass
            self._wakeup.clear()

还记得上面提到的 Scheduler 构造函数中的 self._wakeup 么，它实际上是 threading.Event ，它的 wait 方法会一直 block 直到 Event flag 为 True，也就是我们上面看到的 self._wakeup.set() ，那么我们可以知道在 Scheduler 中有几种场景会置为 True：

Scheduler.shutdown
Scheduler.add_jobstore
Scheduler._real_add_job

如果没有触发上述场景，则 _main_loop 会根据 jobs 的执行时间一直循环等待。

JobStore

在 apscheduler 中，JobStore 只是单纯的实现了 Job 相关的方法：

class JobStore(object):
    def add_job(self, job):
    def update_job(self, job):
    def remove_job(self, job):
    def load_jobs(self):
    def close(self):

其中，对 job 的操作会根据 JobStore 类型的不同，而采用不同的序列化方式，比如在 MongoDBJobStore 中采用的是 bson.binary，而在其他 JobStore 比如 RedisJobStore 中采用的都是 pickle。

Events

在 Scheduler 中，我们已经看到通过 threading.Event 来实现事件通知的，那么我们通知的 Event 都是在 apscheduler.events 中定义好的，比如：

class JobEvent(SchedulerEvent):
    def __init__(self, code, job, scheduled_run_time, retval=None,
                 exception=None, traceback=None):
        SchedulerEvent.__init__(self, code)
        self.job = job
        self.scheduled_run_time = scheduled_run_time
        self.retval = retval
        self.exception = exception
        self.traceback = traceback

在 JobEvent 中，我们能看到 job 的执行时间，返回值，异常捕获等信息。如果看过之前关于 huey 博客的同学应该知道，在 huey 中是可以直接通过 task id 获取 task 执行结果的，但是在 apscheduler 中，我们并没有直接获取该结果的方法，而是通过在 Scheduler 中的 add_listener 添加监听者，监控指定成功的 Job 获取该 Job 的返回值，感觉这里不太友好。

总结

到这里我们基本上已经将 apscheduler 的流程走了一遍，具体的 Trigger 计算时间的方法之后有机会单独写一下关于 cron，interval，date 的计算方法。

与 huey 相比，apscheduler 使用上要简单，但是简单也意味着功能的不足，比如获取 job 执行结果、job retry 机制等等。当然也有比较好的地方，apscheduler 在跟 web 框架比如 Flask，Django 集成的时候有一些第三方插件可以直接使用，不用像 Huey 一样要单独启动一个 consumer 进程，比较方便。

记一次 Python 编码踩坑

发表于 2019-04-02 |

背景

一直知道 Python 容易踩编码的坑，尤其是 Python2，昨天第一次遇到，记一下。

起因

产品中有一个账号关联的功能，需要的参数大概有 host,port,user,passwd 这么几个参数，昨天发现一个环境中账号关联失败，看请求应该还没到账号认证那里就失败了，查看 rest-server 日志，并没有发现错误异常，api 也是正常返回的，通过其他方式验证账号是有效的，当时觉得很奇怪，没什么想法。

调查

既然 rest-server 中日志没有报错，那么看看服务是否有什么异常。
这里特意说名下，如果服务使用的是 gunicorn 或者 celery 等第三方库作为守护进程，有一些系统报错是不会记录到你的服务中的，而是会直接打印到系统中（messages or systemd）。

发现 systemctl status 和 journal -u 有报错，报错内容如下：

4月 02 06:20:00 SCVM70 gunicorn[31225]: Traceback (most recent call last):
4月 02 06:20:00 SCVM70 gunicorn[31225]: File "/usr/lib64/python2.7/site-packages/gevent/threadpool.py", line 207, in _worker
4月 02 06:20:00 SCVM70 gunicorn[31225]: value = func(*args, **kwargs)
4月 02 06:20:00 SCVM70 gunicorn[31225]: error: getaddrinfo() argument 2 must be integer or string
4月 02 06:20:00 SCVM70 gunicorn[31225]: (<ThreadPool at 0x120ae50 0/5/10>, <built-in function getaddrinfo>) failed with error

字面意思是传递参数的类型不对，必须为 int 或者 string。

在 Chrome 中查看当时 API 请求参数，所有参数均为 string，我也实际用 int 或者 str 类型进行验证，账号验证都是可以成功的，感觉进入了死胡同。

原因

最后发现是编码的问题，传递的 port 参数是 unicode 编码，而 getaddrinfo() 需要的是 int 或者 string。

在 python2中，str 其实是 bytes，而不是 unicode，在代码中声明了编码方式为 utf-8，并将该参数存入到了 DB 中，导致下次请求传递的还是 DB 中的 utf-8 类型的 port，而不是 int 或者 string。

解决

这里解决的方式很简单，直接将 port 强制转换为 int 或者 string 就好。根本解决方式应该是在之后的代码编写过程中，规范变量类型，比如 port 就使用 int，而不是 str（unicode），防止出现未知错误。

huey 源码阅读

发表于 2019-03-31 |

背景

最近读完了《Redis 实战》对 Redis 有了一些了解，但是没有在实际项目中应用过，就想找一个使用 Redis 的项目来看看，找到 Huey 是因为之前使用过，趁机了解下具体实现。

简介

Huey 的定位是一个轻量级的任务队列，仅依赖于 Redis 作为任务相关信息存储，支持的功能有：

多种 worker 执行方式：thread，process，greenlet
支持多种任务类型：特定时间运行，周期性运行
包含重试机制，可以指定重试次数及重试间隔
支持任务锁
…

我们根据官方的示例，来看看 Huey 是如何处理任务的，目录结构如下：

master ✔ $ pwd
/Users/yiran/Documents/git-repo/huey/examples/simple
yiran@zhouyirandeMacBook-Pro:~/Documents/git-repo/huey/examples/simple
master ✔ $ tree .
.
├── README
├── __init__.py
├── config.py
├── cons.sh
├── main.py
└── tasks.py

注意，这个目录结构是 Huey 官方建议的，具体原因为：

Behind-the-scenes when you decorate a function with task() or periodic_task(), the function registers itself with a centralized in-memory registry. When that function is called, a reference is put into the queue (along with the arguments the function was called with, etc), and when that message is consumed, the function is then looked-up in the consumer’s registry. Because of the way this works, it is strongly recommended that all decorated functions be imported when the consumer starts up.

Task

Huey 支持通过 @task 装饰器的方式创建任务，示例如下：

from huey import RedisHuey

huey = RedisHuey('simple.test', blocking=True)

@huey.task()
def count_beans(num):
    print('-- counted %s beans --' % num)
    return 'Counted %s beans' % num

从 huey.RedisHuey 创建了 huey 实例，我们看下 RedisHuey 是什么：

class RedisHuey(Huey):
    def get_storage(self, read_timeout=1, max_errors=1000,
                    connection_pool=None, url=None, **connection_params):
        return RedisStorage(
            name=self.name,
            blocking=self.blocking,
            read_timeout=read_timeout,
            max_errors=max_errors,
            connection_pool=connection_pool,
            url=url,
            **connection_params)

在 huey.storage 中定义了 RedisHuey，继承自 Huey 类，重新实现了 get_storage 方法，我们看下 RedisStorage 是做什么用途的：

class RedisStorage(BaseStorage):
    redis_client = Redis

    def __init__(self, name='huey', blocking=False, read_timeout=1,
                 max_errors=1000, connection_pool=None, url=None,
                 client_name=None, **connection_params):

        if Redis is None:
            raise ImportError('"redis" python module not found, cannot use '
                              'Redis storage backend. Run "pip install redis" '
                              'to install.')

        if sum(1 for p in (url, connection_pool, connection_params) if p) > 1:
            raise ValueError(
                'The connection configuration is over-determined. '
                'Please specify only one of the the following: '
                '"url", "connection_pool", or "connection_params"')

        if url:
            connection_pool = ConnectionPool.from_url(
                url, decode_components=True)
        elif connection_pool is None:
            connection_pool = ConnectionPool(**connection_params)
    ...

    def clean_name(self, name):
    def convert_ts(self, ts):
    def enqueue(self, data):
    def dequeue(self):
    ...
    def get_errors(self, limit=None, offset=0):
    def flush_errors(self):
    def emit(self, message):
    def listener(self):
    def __iter__(self):

RedisStorage 继承自 Storage，主要用于所有任务相关信息的写入及读取，目前 Huey 中只实现了 Storage 介质，如果想使用其他方式（如 MongoDB）就需要自己实现了。

知道了 RedisHuey 跟 Huey 的主要区别只是在于 get_storage 方法不同，那么看看 task 方法的具体实现：

def task(self, retries=0, retry_delay=0, retries_as_argument=False,
         include_task=False, name=None, **task_settings):
    def decorator(func):
        """
        Decorator to execute a function out-of-band via the consumer.
        """
        return TaskWrapper(
            self,
            func.func if isinstance(func, TaskWrapper) else func,
            retries=retries,
            retry_delay=retry_delay,
            retries_as_argument=retries_as_argument,
            include_task=include_task,
            name=name,
            **task_settings)
    return decorator

我们看下 TaskWrapper 做了啥，一些变量复制忽略掉，关键的是这几行:

class TaskWrapper(object):
    def __init__(self, huey, func, retries=0, retry_delay=0,
                 retries_as_argument=False, include_task=False, name=None,
                 task_base=None, **task_settings):
        ...
        self.task_class = create_task(
            QueueTask if task_base is None else task_base,
            func,
            retries_as_argument,
            name,
            include_task,
            **task_settings)
        self.huey.registry.register(self.task_class)

    def is_revoked(self, dt=None, peek=True):
        return self.huey.is_revoked(self.task_class, dt, peek)

    def revoke(self, revoke_until=None, revoke_once=False):
        self.huey.revoke_all(self.task_class, revoke_until, revoke_once)

将函数 func 创建为一个任务，并将任务注册到 Huey 中。同时实现了一些方法，这些方法最终调用的都是在 Huey 类中实现的。

Consumer

root@yiran30250:~/project/huey/examples/simple
master ✔ $ cat cons.sh
#!/bin/bash
echo "HUEY CONSUMER"
echo "-------------"
echo "In another terminal, run 'python main.py'"
echo "Stop the consumer using Ctrl+C"
PYTHONPATH=.:$PYTHONPATH
export WORKER_CLASS=${1:-thread}
python ../../huey/bin/huey_consumer.py main.huey --workers=2 -v -s 10 -k $WORKER_CLASS -C

接下来我们看下 Consumer 是如何实现的，在进程启动脚本里，先忽略其他参数，我们指定了 huey 实例，并指定了 worker 数量为 2，我们看下是如何执行的：

huey_consumer.py

def consumer_main():
    ...
    huey_instance = load_huey(args[0])
    config.setup_logger()
    consumer = huey_instance.create_consumer(**config.values)
    consumer.run()
if __name__ == '__main__':
    consumer_main()

加载 huey 实例，创建相应 Consumer 实例并运行，我们看下 Consumer 做了什么：

class Consumer(object):
    """
    Consumer sets up and coordinates the execution of the workers and scheduler
    and registers signal handlers.
    """

根据注释我们可以知道 Consumer 主要是启动了 worker 和 scheduler，并截获相应的信息进行处理。
在 huey/consumer.py 中定义了 Worker,Scheduler,Environment 类，其中 Environment 根据所指定的 worker 类型的不同，分为 thread,process,greenlet。

我们看下 Worker 是如何执行对应的任务的：

Worker

def loop(self, now=None):
    task = None
    exc_raised = True
    try:
        task = self.huey.dequeue() # 从 hue 中获取任务，也就是从 redis 中获取任务
        ...
    if task:
        self.delay = self.default_delay
        self.handle_task(task, now or self.get_now()) # 如果获取到了任务，对任务进行处理
    elif exc_raised or not self.huey.blocking:
        self.sleep()

def handle_task(self, task, ts):
    if not self.huey.ready_to_run(task, ts):
        self.add_schedule(task) # 若任务没有到达执行时间，则添加到 schedule 中
    elif not self.is_revoked(task, ts):
        self.process_task(task, ts) # 若任务没有被取消，则执行任务
    else:
        self.huey.emit_task(
            EVENT_REVOKED,
            task,
            timestamp=ts)
        self._logger.debug('Task %s was revoked, not running', task) # 任务被取消，通知 event

def process_task(self, task, ts):
    ...
    self.run_pre_execute_hooks(task)   # 执行 pre hook 动作
    ...  
    task_value = self.huey.execute(task) # 具体执行 task
    ...
    self.run_post_execute_hooks(task, task_value, exception) # 执行 post hook 动作
    ...

看到 self.huey.execute(task) 我们发现最终还是通过 Huey 中的方法 execute 来执行 task，那么我们看下是如何实现的：

def execute(self, task):
    if not isinstance(task, QueueTask):
        raise TypeError('Unknown object: %s' % task)

    try:
        result = task.execute()

发现这里调用的是 task.execute，这里的 task 是谁呢，是我们最开始提到的 TaskWrapper 么，在 TaskWrapper 中没有实现 execute 方法，这里其实是上面提到的 create_task 返回的类，我们看下 create_task 的具体实现：

def create_task(task_class, func, retries_as_argument=False, task_name=None,
                include_task=False, **kwargs):
    def execute(self):
        args, kwargs = self.data or ((), {})
        if retries_as_argument:
            kwargs['retries'] = self.retries
        if include_task:
            kwargs['task'] = self
        return func(*args, **kwargs)

    attrs = {
        'execute': execute,
        '__module__': func.__module__,
        '__doc__': func.__doc__}
    attrs.update(kwargs)

    if not task_name:
        task_name = func.__name__

    return type(task_name, (task_class,), attrs)

可以看到 execute 函数中最终执行的就是 func 自己， func(*args, **kwargs) ，通过 type 返回真正的 task 类，指定父类为 task_class （默认是 QueueTask）。

到这里我们就完成了整个流程：任务注册，任务调度，任务执行。那么我们再来看看 Huey 中的锁是怎么实现的。

Lock

class TaskLock(object):
    """
    Utilize the Storage key/value APIs to implement simple locking. For more
    details see :py:meth:`Huey.lock_task`.
    """
    def __init__(self, huey, name):
        self._huey = huey
        self._name = name
        self._key = '%s.lock.%s' % (self._huey.name, self._name)
        self._huey._locks.add(self._key)

    def __call__(self, fn):
        @wraps(fn)
        def inner(*args, **kwargs):
            with self:
                return fn(*args, **kwargs)
        return inner

    def __enter__(self):
        if not self._huey._put_if_empty(self._key, '1'):
            raise TaskLockedException('unable to set lock: %s' % self._name)

    def __exit__(self, exc_type, exc_val, exc_tb):
        self._huey._get_data(self._key)

1 2	def put_if_empty(self, key, value): return self.conn.hsetnx(self.result_key, key, value)

在 Huey 中，任务锁的实现非常简单，单纯的利用了下 Redis 中的 HSETNX 机制：

Sets field in the hash stored at key to value, only if field does not yet exist. If key does not exist, a new key holding a hash is created. If field already exists, this operation has no effect.

总结

Huey 作为一个任务调度应用，整体代码量不多，但是有很多值得学习的地方，也许过段时间再看会有新的收获。