网络部署上线过程记录

以推荐相似单品项目为例：reco-similar-product

项目目录

Dockerfile
Jenkinsfile
README.md
config
deploy.sh
deployment.yaml
index
offline
recsys
requirements.txt
service
stat

可以看到，目录中有许多文件，这里简单介绍下各个目录的作用：

• Dockerfile：Docker部署脚本；
• Jenkinsfile：Jenkinsfile 是一个文本文件，它包含了 Jenkins 流水线的定义并被检入源代码控制仓库;
• README.md： 项目介绍；
• config：存放配置文件的位置；
• deploy.sh：在Docker上部署的脚本;
• deployment.yaml：k8-s配置文件；
• index：存放index数据的位置，这里用来做ctr平滑；
• offline：离线部分的代码；
• recsys：存放一些通用调用模块
• requirements.txt：项目所依赖的包;
• service：服务器部分的代码；
• stat：相关统计脚本

详细介绍

Dockerfile

FROM tools/python:3.6.3 

ADD . /usr/local/reco_similar_product_service

WORKDIR /usr/local/reco_similar_product_service

EXPOSE 5001

# build image
RUN mkdir -p /data/reco_similar_product_service/log
RUN bash deploy.sh build

# launch image
CMD bash deploy.sh launch

• from tools/python:3.6.3 ： 从此处获取python镜像包
• ADD . /usr/local/reco_similar_product_service：ADD指令的功能是将主机构建环境（上下文）目录中的文件和目录、以及一个URL标记的文件 拷贝到镜像中。把文件放在该文件中
• WORKDIR /usr/local/reco_similar_product_service：指定工作目录
• EXPOSE 5000：暴露端口号，声明端口
• RUN mkdir -p /data/algoreco_simproduct_service/log：设置日志文件地址 ，与后面的log日志地方保持一致
• RUN bash deploy.sh build：调用deploy.sh部署
• CMD bash deploy.sh launch：启动镜像

Jenkinsfile

node('jenkins-jnlp') {
    stage("Clone") {
        checkout scm
    }
    stage("Tag") {
        script {
            tag = sh(returnStdout: true, script: 'git rev-parse --short HEAD').trim()
        }
    }
    stage("Push Docker Image") {
    sh "sed -i 's@TAG@${tag}@g' deployment.yaml"
        withDockerRegistry(credentialsId: '', url: '') {
            sh "docker build -t reco-similar-product:${tag} ."
            sh "docker push reco-similar-product:${tag}"
        }
    }
    stage("Deployment") {
        sh "kubectl apply -f deployment.yaml"
    }
}

上述是 Jenkins 流水线的几个过程，克隆特定修订版本（Clone），获取tag， 创建镜像（build），将本地的镜像上传到镜像仓库（push），对文件或标准输入流更改资源应用配置（Deployment）

• jenkins-jnlp: Jenkins的Agent大概分两种。 一是基于SSH的，需要把Master的SSH公钥配置到所有的Agent宿主机上去。 二是基于JNLP的，走HTTP协议，每个Agent需要配置一个独特的密码。
• stage('Clone'): stage 指令在 stages 部分进行，应该包含一个 实际上, 流水巷所做的所有实际工作都将封装进一个或多个 stage 指令中。
• checkout scm: checkout 步骤将会从源代码控制中检出代码；scm 是一个特殊的变量， 它指示 checkout 步骤克隆触发流水线运行的特定修订版本;
• docker build和docker push：创建镜像和上传镜像
• kubectl apply -f deployment.yaml： 对文件或标准输入流更改资源应用配置

deploy.sh

#!/bin/bash

# Dependencies in $PATH (in base docker image):
# - python 3.6.3
# - pip for this python


build()
{
  # prepare env if there's no docker
  # ENV_DIR="venv"
  # pip install virtualenv
  # virtualenv --no-site-packages $ENV_DIR
  # source $ENV_DIR/bin/activate
  pip install -r requirements.txt \
    -i http://mirrors.aliyun.com/pypi/simple/ --trusted-host mirrors.aliyun.com
  python setup.py build_ext --inplace
}


launch()
{
  # start web container
  cd service/
  bash run.sh start
}


if [ x"$1" == "xbuild" ]; then
  echo "build"
  build
fi

if [ x"$1" == "xlaunch" ]; then
  echo "launch"
  launch

• build: 安装相应的依赖包；
• launch： 启动服务；

deployment.yaml

apiVersion: extensions/v1beta1
kind: Deployment
metadata:
  labels:
    k8s-app: reco-similar-product-dev
  name: reco-similar-product-dev
  namespace: dev
spec:
  minReadySeconds: 5
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxSurge: 25%
      maxUnavailable: 25%
  replicas: 1
  selector:
    matchLabels:
      k8s-app: reco-similar-product-dev
  template:
    metadata:
      labels:
        k8s-app: reco-similar-product-dev
    spec:
      serviceAccountName: dev
      nodeSelector:
        environment: dev
      imagePullSecrets:
      - name: dev
      containers:
      - name: reco-similar-product-dev
        image: develop/reco-similar-product:TAG
        env:
        - name: RECO_SIM_TEST
          value: "1"
        ports:
        - containerPort: 5000
          protocol: TCP
        volumeMounts:
        - name: reco-similar-product-dev-log
          subPath: reco-similar-product-dev-log
          mountPath: /data/reco_similar_product_service/log
      securityContext:
        fsGroup: 1000
      volumes:
      - name: reco-similar-product-dev-log
        persistentVolumeClaim:
          claimName: dev-logs
---
kind: Service
apiVersion: v1
metadata:
  labels:
    k8s-app: reco-similar-product-dev
  name: reco-similar-product-dev
  namespace: dev
spec:
  ports:
  - port: 5000
    targetPort: 5000
  selector:
    k8s-app: reco-similar-product-dev

• dev： 表示测试环境
• port： 端口号
• app： 服务名
• mountPath：日志文件路径

config

该目录主要是存放一些数据库，路径配置等配置文件

recsys

放置一些通用模块，比如基础模块：数据库连接等模块， 服务端的核心处理逻辑代码也可以放在这里

offline

离线部分代码，主要是服务功能的主逻辑部分代码。对于推荐相似单品而言，就是要去生成相应的数据，写入到表里。代码的作用是，处理数据，离线计算处理得到一份结果，将结果写入到数据库中。

#!/usr/bin/env python
# fileUsing: reco some pid from sale's pids
# retrieve pids by cv feature
# filt by type, alive

import os
import sys
import oss2
import json
import yiss
import numpy as np
from operator import itemgetter
from datetime import datetime


MODULE_REAL_DIR = os.path.dirname(os.path.realpath(__file__))
DATA_PATH = MODULE_REAL_DIR + '/../data'
CONFIG_PATH = '../../../config/'


class SearchItem():
    def __init__(self, feat=None, pid=None, p_type=None, score=None):
        '''
        create item struct
        every pid contain:
            feat: cv feature
            pid: product id
            p_type: type 
            score: similar score
        '''
        self.feat = feat
        self.pid = pid
        self.p_type = p_type
        self.score = score


class ImgSearch(SearchItem):
    def __init__(self):
        '''
        retrieve mode:
            ANN: Approximate Nearest Neighbor
        '''

        # prepare cv feature dataset
        self.product_feature = {}
        feature_path = DATA_PATH + '/product_cv_feature.json'
        with open(feature_path, 'r') as f:
            for line in f:
                pid, _, feat = line.rstrip('\n').split('\t')
                self.product_feature[int(pid)] = json.loads(feat)  # change key type into int

        # -> init build index and save index <-
        self.index = yiss.VectorIndex(
            data_file=feature_path, series='IDMap, Flat', suff_len=1, search_width=1)
        self.index.init(save_file=MODULE_REAL_DIR + '/../data/vector.index')

        # init ImageUtils
        self.ImgUtil = yiss.ImageUtils(CONFIG_PATH + '/config.json')

        # load tags: type, alive, color
        tag_path = DATA_PATH + "/pid_tags"
        self.pid_type = {}
        self.pid_alive = {}
        with open(tag_path, 'r') as f:
            for line in f:
                pid, type_top_id, is_alive = line.rstrip('\n').split('\t')
                self.pid_type[pid] = type_top_id
                self.pid_alive[pid] = is_alive

        # init retrieve length
        self.RETRIEVE_LEN = 1000

        # init table length
        self.TABLE_LEN = 64

    def retrieve(self, input_item):
        """
        search similar feature pids by faiss
        """
        similar_items = []
        if input_item.feat == []:
            return similar_items  # no feat, could not compare
        feat = np.array([input_item.feat.tolist()])
        feat = feat.astype(np.float32)  # change feat into the faiss needed type
        ds, ids = self.index.knn_search(feat, self.RETRIEVE_LEN)
        for pid, distance in zip(ids, ds):
            if pid != str(input_item.pid):  # ignore itself
                similar_items.append({"pid": pid, "score": round(distance, 4)})
        return sorted(similar_items, key=lambda e: e.__getitem__('score'))

    def build_item(self, pid):
        feat = np.array(self.product_feature[pid]) if pid in self.product_feature else []
        pid = str(pid)  # pid type and color dict key type is <string>
        pid_type = self.pid_type[pid] if pid in self.pid_type else self.ImgUtil.get_type_id(pid)
        return SearchItem(feat, int(pid), pid_type)

    def filter_pid(self, input_item, retrieve_items):
        """
        filter by type, alive
        """
        match_result = []
        not_match_result = []
        finally_match_result = []
        input_pid = input_item.pid
        input_type = input_item.p_type
        for item in retrieve_items:
            pid, score = str(item["pid"]), item["score"]  # change pid type from int into string
            p_type = self.pid_type[pid] if pid in self.pid_type else\
                self.ImgUtil.get_type_id(pid)
            p_alive = self.pid_alive[pid] if pid in self.pid_alive else "0"
            item = {"pid": int(pid), "score": score, "path": "sVSM"}
            if p_alive == "1":
                if input_type == p_type:
                    match_result.append(item)
                    if len(match_result) >= self.TABLE_LEN:
                        return match_result
                finally_result = match_result
        return finally_result[:self.TABLE_LEN]

    def search(self, pid):
        """
        main function, input pid and search similar pids
        """
        pid = int(pid)
        # build input item
        pid_item = self.build_item(pid)
        # print("pid item:", pid_item.pid, pid_item.p_type, pid_item.feat)
        if len(pid_item.feat) == 0:  # feat is null, ignore this pid
            return []
        # retrieve
        retrieve_item = self.retrieve(pid_item)
        # filter by some condition
        match_result = self.filter_pid(pid_item, retrieve_item)
        return match_result


if __name__ == "__main__":
    img_search = ImgSearch()
    pid = sys.argv[1]
    match_result = img_search.search(pid)
    if len(match_result) > 0:
        match_result_pid = [int(item["pid"]) for item in match_result]
        print(match_result_pid)

这里先用Ann依照特征相似度进行召回，然后根据类别进行过滤，最后将结果保存下来。获得结果后上传到数据库中。

service

服务端目录

• gconfig.py： gunicorn配置文件
• recoservice.py： 使用flask提供线上服务接口
• run.sh： 启动或关停服务
• ping.sh：

gconfig.py :

import multiprocessing

bind = '0.0.0.0:5001'  # 绑定ip和端口号
daemon = False
debug = False
worker_class = 'sync'
workers = multiprocessing.cpu_count()  # 进程数
pidfile = 'project.pid'
log_path = "/data/algoreco_simproduct_service/log/"  # 日志文件 保证与之前的位置一样
timeout = 300
graceful_timeout = 300

# http://docs.gunicorn.org/en/stable/settings.html
access_log_format = '%(h)s %(l)s %(u)s %(t)s "%(r)s" %(s)s %(b)s "%(f)s" "%(a)s" "%({xff}i)s"'
logconfig_dict = {
    'version': 1,
    'disable_existing_loggers': False,
    'loggers': {
        "gunicorn.error": {
            "level": "INFO",
            "handlers": ["error_file"],
            "propagate": 0,
            "qualname": "gunicorn.error"
        },
        "gunicorn.access": {
            "level": "INFO",
            "handlers": ["access_file"],
            "propagate": 0,
            "qualname": "gunicorn.access"
        }
    },
    'handlers': {
        "error_file": {
            "class": "cloghandler.ConcurrentRotatingFileHandler",
            "maxBytes": 1024 * 1024 * 1024,
            "backupCount": 5,
            "formatter": "generic",
            "filename": log_path + "gunicorn_error.log",
        },
        "access_file": {
              "level": "INFO",
            "handlers": ["access_file"],
            "propagate": 0,
            "qualname": "gunicorn.access"
        }
    },
    'handlers': {
        "error_file": {
            "class": "cloghandler.ConcurrentRotatingFileHandler",
            "maxBytes": 1024 * 1024 * 1024,
            "backupCount": 5,
            "formatter": "generic",
            "filename": log_path + "gunicorn_error.log",
        },
        "access_file": {
            "class": "cloghandler.ConcurrentRotatingFileHandler",
            "maxBytes": 1024 * 1024 * 1024,
            "backupCount": 5,
            "formatter": "generic",
            "filename": log_path + "gunicorn_access.log",
        }
    },
    'formatters': {
        "generic": {
            "format": "%(asctime)s [%(process)d] %(levelname)s"
                      + " [%(filename)s:%(lineno)s] %(message)s",
            "datefmt": "[%Y-%m-%d %H:%M:%S]",
            "class": "logging.Formatter"
        },
        "access": {
            "format": "[%(asmidnighttime)s] [%(process)d] %(levelname)s"
                      + " [%(filename)s:%(lineno)s] %(message)s",
            "class": "logging.Formatter"
        }
    }
}

• worker_class： 工作进程类型, 有sync（默认）, eventlet, gevent, or tornado, gthread, gaiohttp；
• workers： 工作进程的数量，通常推荐的worker数量是：(2 x $num_cores) + 1
• log-level LEVEL: 输出error log的颗粒度，有效的LEVEL有: debug, info, warning, error, critical

recoservice.py :

import logging
import json
from flask import Flask, Response, request
from recsys.sale.reco import RecoSaleSim
from recsys.base.exception import ClientException
from recsys.base.recologging import HeartBeatFilter

app = Flask(__name__)


class RecoSimSvr():
    def __init__(self):
        self.SUCCESS_CODE = 200
        self.CLIENT_ERROR_CODE = 400
        self.SERVER_ERROR_CODE = 500

        # init recsys module
        self.rss = RecoSaleSim()
        app.logger.info('Init done.')

    def get_recos(self):
        error_results = {}
        results = {}
        try:
            pid = request.args.get('productId', type=int)
            if not pid:
                raise ClientException('Param missing: productId')
            uid = request.args.get('uid', type=int)
            if not uid:
                raise ClientException('Param missing: uid')
            device_id = request.args.get('deviceId', type=str)
            if not device_id:
                raise ClientException('Param missing: deviceId')
            reco_len = request.args.get('maxLength', type=int, default=16)
            reco_items = self.rss.get_reco_items(pid, reco_len)
            # Output response
            results['data'] = reco_items
            results['code'] = self.SUCCESS_CODE
            results['msg'] = 'Success'
            return Response(json.dumps(results), status=200, mimetype='application/json')
        except ClientException as err:
                    app.logger.error(str(err), exc_info=True)
            error_results['code'] = self.CLIENT_ERROR_CODE
            error_results['msg'] = str(err)
            return Response(json.dumps(error_results), status=400, mimetype='application/json')
        except Exception as err:
            app.logger.error(str(err), exc_info=True)
            error_results['code'] = self.SERVER_ERROR_CODE
            error_results['msg'] = 'Program error!'
            return Response(json.dumps(error_results), status=500, mimetype='application/json')

    def check_alive(self):
        results = {}
        results['code'] = 200
        results['msg'] = 'I\'m Sale Similar Reco'
        return Response(json.dumps(results), status=200, mimetype='application/json')


# setup logger
gunicorn_access_logger = logging.getLogger('gunicorn.access')
gunicorn_access_logger.addFilter(HeartBeatFilter())
app.logger.handlers = gunicorn_access_logger.handlers
app.logger.setLevel(gunicorn_access_logger.level)

# init main class
reco_sim_svr = RecoSimSvr()

# add url rule
app.add_url_rule('/simproduct/', view_func=reco_sim_svr.check_alive, methods=['GET'])
app.add_url_rule('/simproduct/getSaleSimRecos', view_func=reco_sim_svr.get_recos, methods=['GET'])
app.logger.info('All Init Done.')

run.sh:

#!/bin/bash

boot_dir=../

start()
{
  gunicorn --chdir $boot_dir --config gconfig.py service.recoservice:app
}


stop()
{
  pid_file=${boot_dir}/project.pid
  if [ ! -f $pid_file ];then
    echo "Cannot find pid file project.pid. Maybe the service is not running."
    return 255
  fi
  pid=`cat $pid_file`
  kill -15 $pid
}


if [ $1"x" == "startx" ];then
  start
fi

if [ $1"x" == "stopx" ];then
  stop
fi

if [ $1"x" == "restartx" ];then
  stop
  echo "wait for stopping..."
  sleep 5
  start
fi

ping.sh:

#!/usr/bin/env bash

if [ $# -lt 1 ]; then
    echo "Usage: bash $0 {online|mock|test|single|local}"
    exit -1
fi

SENV=$1
BASE_DOMAIN=xxxx.com
if [ $SENV"x" == "testx" ]; then
    DOMAIN="test"$BASE_DOMAIN
elif [ $SENV"x" == "onlinex" ]; then
    DOMAIN=$BASE_DOMAIN
elif [ $SENV"x" == "singlex" ]; then
    DOMAIN=x.x.x.x
    PORT=5001
elif [ $SENV"x" == "localx" ]; then
    DOMAIN=0.0.0.0
    PORT=5001
elif [ $SENV"x" == "mockx" ]; then
    DOMAIN="mock"$BASE_DOMAIN
fi

device='TEST85EA5969-4398-4A78-8816-348E189'


run_once()
{
  curl -X GET "http://$DOMAIN:$PORT/simproduct/"; echo
  curl -X GET "http://$DOMAIN:$PORT/simproduct/getSaleSimRecos?uid=1&deviceId=$device&productId=51953&maxLength=16"; echo
}


basic_test()
{
  # test if success
  echo "test service..."
  run_once
}


basic_test

stat

统计脚本放的相关代码

参考

• 如何编写最佳的Dockerfile
• Docker--从入门到实践
• 使用Jenkins
• docker 教程
• Gunicorn运行与配置