阿里云证书过期时间监测

我只监控了slb下使用的ssl证书的过期时间，然后以prometheus能读取的metrics格式供prometheus采集。因为我们是prometheus-operator,通过helm发布的时候添加一个servicemonitor即可。想要监控所有证书调用阿里云证书系列产品api即可。

#!/usr/bin/python
# -*- coding: UTF-8 -*-

'''
Date: 2019-11-26
Blog: www.cnblogs.com/qinghe123
'''

import re,sys
import base64
import json
import time
from datetime import datetime
import hmac
import urllib

import argparse
from pprint import pprint
from urllib import request
import uuid
import os
from sys import exit

import dateutil.parser
from prometheus_client import start_http_server, Summary
from prometheus_client.core import GaugeMetricFamily, REGISTRY
from hashlib import sha1

DEBUG = int(os.environ.get('DEBUG', '0'))
#DEBUG = True
aliyun_access_key = os.environ.get('AccessKeyID',None)
aliyun_access_secret = os.environ.get('AccessKeySecret',None)

COLLECTION_TIME = Summary('request_aliyun_api_seconds', 'Time spent to collect metrics from aliyun')


class AliyunSignature():
    """获取阿里云签名和数据"""
    Timestamp = datetime.utcnow().isoformat()
    SignatureNonce = str(uuid.uuid1())

    def __init__(self,config,Id=None,key=None):
        '''
            this is aliyun api auth single, now, any api request parame to main
        '''
        assert config['Action'], "Value error"

        self.config = config
        self.url = 'http://slb.aliyuncs.com'
        self.id = Id
        self.Key = key
        self.__data = dict({
            "AccessKeyId": self.id,
            "SignatureMethod": 'HMAC-SHA1',
            "SignatureVersion": "1.0",
            "SignatureNonce": self.SignatureNonce,
            "Timestamp": self.Timestamp,
            "Version": "2014-05-15",
            "Format": "JSON"
        }, **self.config)

    @classmethod
    def reload(cls, body):
        cls.Timestamp = datetime.utcnow().isoformat()
        cls.SignatureNonce = str(uuid.uuid1())
        return cls(config=body)

    @property
    def data(self):
        return self.__data

    @data.setter
    def data(self, value):
        if self.__data:
            raise AssertionError("not allow opeartion")
        self.__data = value

    @staticmethod
    def percent_encode(encodeStr):
        if isinstance(encodeStr, bytes):
            encodeStr = encodeStr.decode(sys.stdin.encoding)
        res = urllib.parse.quote_plus(encodeStr.encode('utf-8'), '')
        res = res.replace('+', '%20').replace('*', '%2A').replace('%7E', '~')
        return res

    def auth(self):
        base = sorted(self.data.items(), key=lambda data: data[0])
        canstring = ''
        for k, v in base:
            canstring += '&' + self.percent_encode(k) + '=' + self.percent_encode(v)
        self.Key += "&"
        data = 'GET&%2F&' + self.percent_encode(canstring[1:])
        self._salt(data)
        return self.data

    def _salt(self, data):
        result = data.encode(encoding='utf-8')
        uri = hmac.new(self.Key.encode("utf-8"), result, sha1).digest()
        key = base64.b64encode(uri).strip()
        self.data['Signature'] = key
        return self.data

    def generate_url(self):
        parameters = self.auth()
        url = self.url + '/?' + urllib.parse.urlencode(parameters)
        return url

    def do_action(self):
        url = self.generate_url()
        request = urllib.request.Request(url)
        try:
            conn = urllib.request.urlopen(request)
            response = conn.read().decode()
        except  urllib.error.HTTPError as e:
            print(e.read().strip())
            raise SystemExit(e)

        try:
            res = json.loads(response)
        except ValueError as e:
            raise SystemExit(e)
        return res

class AliyunResourceConllector():
    """prometheus metrics"""
    def __init__(self,product=None):
        self.prodct = product

    def format_metric_name(self):
        return 'aliyun_%s' % self.prodct

    def collect(self):
        config = {
            "Action": "DescribeServerCertificates",
            "RegionId": "cn-shanghai"
        }
        auth_obj = AliyunSignature(config=config, Id=aliyun_access_key, key=aliyun_access_secret)
        auth_obj.reload(config)
        response = auth_obj.do_action().get('ServerCertificates').get('ServerCertificate')

        sub_metric_list = ["CreateTime","ExpireTime"]
        for instance in response:
            for sub_metric in sub_metric_list:
                snake_case = re.sub('([A-Z])', '_\1', sub_metric).lower()
                date_time = self.getDateTimeFromISO8601String(instance.get(sub_metric))
                uninx_time = int(time.mktime(date_time.timetuple()))
                gauge = GaugeMetricFamily(self.format_metric_name() + snake_case, 'aliyun ssl certificate',labels=['domainname','sslname'])
                gauge.add_metric([instance.get('CommonName'),instance.get('ServerCertificateName')],uninx_time)
                yield gauge

    def _info(self,obj):
        config = {
            "Action": "DescribeServerCertificates",
            "RegionId": "cn-shanghai"
        }

        response = obj.do_action().get('ServerCertificates').get('ServerCertificate')
        obj.reload(config)
        return response

    @staticmethod
    def getDateTimeFromISO8601String(s):
        d = dateutil.parser.parse(s)
        return d

def parse_args():
    parser = argparse.ArgumentParser(
        description='jenkins exporter args jenkins address and port'
    )
    parser.add_argument(
        '-k', '--keyid',
        metavar='key',
        required=False,
        help='aliyun access key Id',
        default=os.environ.get('AccessKeyID', 'LTAIdmc9wg0MxRD7')
    )
    parser.add_argument(
        '-s', '--secret',
        metavar='secret',
        required=False,
        help='aliyun access key secret',
        default=os.environ.get('AccessKeySecret')
    )
    parser.add_argument(
        '-p',
        metavar='project',
        required=False,
        help='aliyun project type(ecs,slb,ssl...)',
        default=os.environ.get('PROJECT_TYPE','ssl')
    )
    parser.add_argument(
        '-r',
        metavar='RegionId',
        required=False,
        help='aliyun resource region(cn-hangzhou,cn-shanghai...)',
        default=os.environ.get('REGION_ID','cn-shanghai')
    )

    return parser.parse_args()

if __name__ == "__main__":
    config = {
        "Action": "DescribeServerCertificates",
        "RegionId": "cn-shanghai"
    }
    collector = AliyunResourceConllector('ssl')
    REGISTRY.register(collector)
    print("Polling . Serving at port: 8000")
    start_http_server(8000)
    while True:
        time.sleep(1)