-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathmain.py
35 lines (31 loc) · 1.56 KB
/
main.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
from kubernetes import client, config, watch
from http.server import BaseHTTPRequestHandler, HTTPServer
from os import environ
metric_port = environ.get('METRIC_PORT', '9102')
metric_path = environ.get('METRIC_PATH', '/metrics')
deploy_source_name = environ.get('DEPLOY_SOURCE_NAME')
deploy_source_namespace = environ.get('DEPLOY_SOURCE_NAMESPACE')
deploy_target_name = environ.get('DEPLOY_TARGET_NAME')
deploy_target_namespace = environ.get('DEPLOY_TARGET_NAMESPACE')
def metric_value():
try:
config.load_incluster_config()
except:
print("ERROR: Incluster config load failed. Loading local kube config. If you see this in cluster check Pod permissions")
config.load_kube_config()
v1 = client.AppsV1Api()
deploy_source = v1.list_namespaced_deployment(deploy_source_namespace, field_selector = f"metadata.name={deploy_source_name}")
deploy_target = v1.list_namespaced_deployment(deploy_target_namespace, field_selector = f"metadata.name={deploy_target_name}")
return deploy_source.items[0].status.replicas / deploy_target.items[0].status.replicas
class handler(BaseHTTPRequestHandler):
def do_GET(self):
if self.path == metric_path:
self.send_response(200)
self.send_header('Content-type','text/html')
self.end_headers()
message = "synchro_scale_coefficient" + " " + str(metric_value())
self.wfile.write(bytes(message, "utf8"))
else:
self.send_response(404)
with HTTPServer(('', int(metric_port)), handler) as server:
server.serve_forever()