-
Notifications
You must be signed in to change notification settings - Fork 1
/
worker_k8s_job_watch.py
40 lines (29 loc) · 1.09 KB
/
worker_k8s_job_watch.py
1
2
3
4
5
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
#!/usr/bin/env python3
# Copyright (C) 2022-Present Indoc Systems
#
# Licensed under the GNU AFFERO GENERAL PUBLIC LICENSE,
# Version 3.0 (the "License") available at https://www.gnu.org/licenses/agpl-3.0.en.html.
# You may not use this file except in compliance with the License.
from common import configure_logging
from kubernetes import client
from kubernetes import config
from kubernetes.config import ConfigException
from config import get_settings
from pipelinewatch.stream_watcher import StreamWatcher
def k8s_init():
try:
config.load_incluster_config()
except ConfigException:
config.load_kube_config()
return client.Configuration()
def get_k8s_batchapi(configuration):
return client.BatchV1Api(client.ApiClient(configuration))
def main():
k8s_configurations = k8s_init()
batch_api_instance = get_k8s_batchapi(k8s_configurations)
settings = get_settings()
configure_logging(settings.LOGGING_LEVEL, settings.LOGGING_FORMAT)
stream_watch = StreamWatcher(batch_api_instance, settings)
stream_watch.run()
if __name__ == '__main__':
main()