oceanbase

1
# -*- coding: utf-8 -*-
2
import copy
3
import os
4
import sys
5
import traceback
6
import time
7
import json
8
import requests
9
from enum import Enum
10
from http import HTTPStatus
11

12
from art import text2art
13

14
OUTPUT = {}
15
RESULT_FILE_KEY = "farm/ob_results/"
16
TASK_QUEUE_FILE_KEY = "farm/ob_jobs/{}.json"
17

18

19
def _range(start, last):
20
    def to_str(pos):
21
        if pos is None:
22
            return ''
23
        else:
24
            return str(pos)
25

26
    return to_str(start) + '-' + to_str(last)
27

28

29
def _make_range_string(range):
30
    if range is None:
31
        return ''
32

33
    start = range[0]
34
    last = range[1]
35

36
    if start is None and last is None:
37
        return ''
38

39
    return 'bytes=' + _range(start, last)
40

41

42
class OssProxy:
43

44
    def __init__(self, endpoint=""):
45
        self.endpoint = endpoint
46

47
    def get_object(self, key, _range=None):
48
        url = "{}/{}".format(self.endpoint, key)
49
        headers = {}
50
        if _range is not None:
51
            _range = (_range, None)
52
            headers.update({"range": _make_range_string(_range)})
53
        res = requests.get(url, headers=headers)
54
        if res.status_code < 400:
55
            result = res.content.decode()
56
            return result
57
        return ""
58

59
    def get_object_meta(self, key):
60
        url = "{}/{}".format(self.endpoint, key)
61
        headers = {}
62
        res = requests.head(url, headers=headers)
63
        return res.headers
64

65
    def exists_object(self, key):
66
        ...
67

68

69
class GithubProxy:
70

71
    def __init__(self, host="api.github.com"):
72
        self.host = host
73

74
    def get_job_by_id(self, project, pipeline_id):
75
        url = "https://{}/repos/{}/actions/runs/{}".format(
76
            self.host, project, pipeline_id
77
        )
78
        try:
79
            res = requests.get(
80
                url, headers={
81
                    "Accept": "application/vnd.github+json"
82
                }
83
            )
84
            status_code = res.status_code
85
            if status_code == HTTPStatus.NOT_FOUND:
86
                return {}
87
            return res.json()
88
        except:
89
            traceback.print_exc()
90
            return {}
91

92

93
class TaskStatusEnum(Enum):
94
    submitting = 0
95
    pending = 1
96
    running = 2
97
    stopping = 3
98
    success = 4
99
    fail = -1
100
    kill = -2
101
    timeout = -3
102
    submit_task_fail = -4
103

104

105
def request(method, url, params=None, payload=None, timeout=10, data=None, without_check_status=False):
106
    params = params or {}
107
    try:
108
        response = requests.request(
109
            method,
110
            url,
111
            params=params,
112
            json=payload,
113
            data=data,
114
            timeout=timeout
115
        )
116
        if not without_check_status and response.status_code >= 300:
117
            try:
118
                msg = response.json()["msg"]
119
            except:
120
                msg = response.text
121
            print("[ERROR] MSG:{}".format(msg))
122
            exit(1)
123
        return response
124
    except Exception:
125
        import traceback
126
        traceback.print_exc()
127
        print("Please contact the management personnel for assistance !")
128
        if not without_check_status:
129
            exit(1)
130

131

132
def monitor_tasks(oss_proxy: OssProxy, github_pipeline_id, timeout):
133
    end_time = time.time() + int(timeout)
134
    end_task = False
135
    while time.time() <= end_time:
136
        if end_task is True:
137
            pass
138
        task_data = get_task_res(oss_proxy, github_pipeline_id)
139
        if task_data:
140
            end_task = True
141

142
        time.sleep(1)
143
        if task_data is not None:
144
            task_status = int(task_data["status"])
145
            if task_status <= TaskStatusEnum.fail.value:
146
                print(TaskStatusEnum._value2member_map_[task_status])
147
                print("there is the output url: {}".format(
148
                    "https://ce-farm.oceanbase-dev.com/farm2/ci/?id={}".format(task_data["task_id"])))
149
                return False
150
            elif task_status >= TaskStatusEnum.success.value:
151
                print(TaskStatusEnum._value2member_map_[task_status])
152
                print("there is the output url: {}".format(
153
                    "https://ce-farm.oceanbase-dev.com/farm2/ci/?id={}".format(task_data["task_id"])))
154
                return True
155

156
        time.sleep(5)
157
    else:
158
        ...
159

160

161
def get_task_res(oss_proxy: OssProxy, github_pipeline_id):
162
    try:
163
        result_key = RESULT_FILE_KEY + "{}.json".format(github_pipeline_id)
164
        origin_task_data = oss_proxy.get_object(result_key)
165
        return json.loads(origin_task_data)
166
    except:
167
        return
168

169

170
def main(pipeline_id, project, timeout):
171
    print("create a new task")
172
    print("working....")
173
    logo = text2art('OceanBase  Farm')
174
    print(logo)
175
    oss_proxy = OssProxy("https://obfarm-ce.oss-cn-hongkong.aliyuncs.com")
176
    github_proxy = GithubProxy()
177
    job_info = github_proxy.get_job_by_id(project, pipeline_id)
178
    attempt_number = job_info["run_attempt"]
179
    run_pipeline_id = "{}-{}".format(pipeline_id, attempt_number)
180
    result = monitor_tasks(oss_proxy, run_pipeline_id, timeout)
181
    if not result:
182
        exit(1)
183

184

185
if __name__ == "__main__":
186
    print(sys.argv)
187
    if len(sys.argv) < 4:
188
        print("Missing relevant parameters !")
189
        OUTPUT.update({"success": -1})
190
        sys.exit(1)
191
    main(sys.argv[1], sys.argv[2], sys.argv[3])
192