From 860062abeb593d5c3b7b8bd2a0cde7be9766accb Mon Sep 17 00:00:00 2001 From: danielsagi Date: Tue, 27 Aug 2019 10:54:08 +0300 Subject: [PATCH] Added Metrics Server Discovery - Distinct from Api Server (#167) * added basic metrics server discovery * improved discovery, and added KNOWN PORTS usage * improved apiserver decision * fixed bug with comparison of IP addresses in kubeservicehost * improved description of api server discovery * added checks with auth_token on discovery * fixed bug in version requests and added to tests * added an abstract 'unrecognized API' event, and a filter for it for classification * changed filtering to be done on the same event * fixed verify on session and removed unnecessary enum * minor changes to comments * added detailed explanation --- src/modules/discovery/apiserver.py | 104 ++++++++++++++++++++++++----- tests/discovery/test_apiserver.py | 21 +++--- 2 files changed, 99 insertions(+), 26 deletions(-) diff --git a/src/modules/discovery/apiserver.py b/src/modules/discovery/apiserver.py index 48c5618..afaf3e1 100644 --- a/src/modules/discovery/apiserver.py +++ b/src/modules/discovery/apiserver.py @@ -1,41 +1,109 @@ +import json import requests import logging from ...core.types import Discovery from ...core.events import handler -from ...core.events.types import OpenPortEvent, Service, Event +from ...core.events.types import OpenPortEvent, Service, Event, EventFilterBase + +KNOWN_API_PORTS = [443, 6443, 8080] + +class K8sApiService(Service, Event): + """A Kubernetes API service""" + def __init__(self, protocol="https"): + Service.__init__(self, name="Unrecognized K8s API") + self.protocol = protocol class ApiServer(Service, Event): """The API server is in charge of all operations on the cluster.""" - def __init__(self, protocol="https"): + def __init__(self): Service.__init__(self, name="API Server") - self.protocol=protocol + +class MetricsServer(Service, Event): + """The Metrics server is in charge of providing resource usage metrics for pods and nodes to the API server.""" + def __init__(self): + Service.__init__(self, name="Metrics Server") -# Other devices could have this port open, but we can check to see if it looks like a Kubernetes node -# A Kubernetes API server will respond with a JSON message that includes a "code" field for the HTTP status code -@handler.subscribe(OpenPortEvent, predicate=lambda x: x.port==443 or x.port==6443 or x.port==8080) -class ApiServerDiscovery(Discovery): - """API Server Discovery - Checks for the existence of a an API Server +# Other devices could have this port open, but we can check to see if it looks like a Kubernetes api +# A Kubernetes API service will respond with a JSON message that includes a "code" field for the HTTP status code +@handler.subscribe(OpenPortEvent, predicate=lambda x: x.port in KNOWN_API_PORTS) +class ApiServiceDiscovery(Discovery): + """API Service Discovery + Checks for the existence of K8s API Services """ def __init__(self, event): self.event = event - + self.session = requests.Session() + self.session.verify = False + def execute(self): - logging.debug("Attempting to discover an API server on {}:{}".format(self.event.host, self.event.port)) - self.make_request(protocol="https") - self.make_request(protocol="http") + logging.debug("Attempting to discover an API service on {}:{}".format(self.event.host, self.event.port)) + protocols = ["http", "https"] + for protocol in protocols: + if self.has_api_behaviour(protocol): + self.publish_event(K8sApiService(protocol)) - def make_request(self, protocol): + def has_api_behaviour(self, protocol): try: - r = requests.get("{}://{}:{}".format(protocol, self.event.host, self.event.port), verify=False) - if ('k8s' in r.text) or ('"code"' in r.text and r.status_code is not 200): - self.event.role = "Master" - self.publish_event(ApiServer(protocol=protocol)) + r = self.session.get("{}://{}:{}".format(protocol, self.event.host, self.event.port)) + if ('k8s' in r.text) or ('"code"' in r.text and r.status_code is not 200): + return True except requests.exceptions.SSLError: logging.debug("{} protocol not accepted on {}:{}".format(protocol, self.event.host, self.event.port)) except Exception as e: logging.debug("{} on {}:{}".format(e, self.event.host, self.event.port)) + +# Acts as a Filter for services, In the case that we can classify the API, +# We swap the filtered event with a new corresponding Service to next be published +# The classification can be regarding the context of the execution, +# Currently we classify: Metrics Server and Api Server +# If running as a pod: +# We know the Api server IP, so we can classify easily +# If not: +# We determine by accessing the /version on the service. +# Api Server will contain a major version field, while the Metrics will not +@handler.subscribe(K8sApiService) +class ApiServiceClassify(EventFilterBase): + """API Service Classifier + Classifies an API service + """ + def __init__(self, event): + self.event = event + self.classified = False + self.session = requests.Session() + self.session.verify = False + # Using the auth token if we can, for the case that authentication is needed for our checks + if self.event.auth_token: + self.session.headers.update({"Authorization": "Bearer {}".format(self.event.auth_token)}) + + def classify_using_version_endpoint(self): + """Tries to classify by accessing /version. if could not access succeded, returns""" + try: + r = self.session.get("{}://{}:{}/version".format(self.event.protocol, self.event.host, self.event.port)) + versions = r.json() + if 'major' in versions: + if versions.get('major') == "": + self.event = MetricsServer() + else: + self.event = ApiServer() + except Exception as e: + logging.error("Could not access /version on API service: {}".format(e)) + + def execute(self): + # if running as pod + if self.event.kubeservicehost: + # if the host is the api server's IP, we know it's the Api Server + if self.event.kubeservicehost == str(self.event.host): + self.event = ApiServer() + else: + self.event = MetricsServer() + # if not running as pod. + else: + self.classify_using_version_endpoint() + + # If some check classified the Service, + # the event will have been replaced. + return self.event \ No newline at end of file diff --git a/tests/discovery/test_apiserver.py b/tests/discovery/test_apiserver.py index 9e7322b..cd4120c 100644 --- a/tests/discovery/test_apiserver.py +++ b/tests/discovery/test_apiserver.py @@ -1,7 +1,7 @@ import requests_mock import time -from src.modules.discovery.apiserver import ApiServer, ApiServerDiscovery +from src.modules.discovery.apiserver import ApiServer, ApiServiceDiscovery from src.core.events.types import Event from src.core.events import handler @@ -13,20 +13,21 @@ def test_ApiServer(): with requests_mock.Mocker() as m: m.get('https://mockOther:443', text='elephant') m.get('https://mockKubernetes:443', text='{"code":403}', status_code=403) + m.get('https://mockKubernetes:443/version', text='{"major": "1.14.10"}', status_code=200) e = Event() e.protocol = "https" e.port = 443 e.host = 'mockOther' - a = ApiServerDiscovery(e) + a = ApiServiceDiscovery(e) a.execute() e.host = 'mockKubernetes' a.execute() # Allow the events to be processed. Only the one to mockKubernetes should trigger an event - time.sleep(0.1) + time.sleep(1) assert counter == 1 def test_ApiServerWithServiceAccountToken(): @@ -35,6 +36,7 @@ def test_ApiServerWithServiceAccountToken(): with requests_mock.Mocker() as m: m.get('https://mockKubernetes:443', request_headers={'Authorization':'Bearer very_secret'}, text='{"code":200}') m.get('https://mockKubernetes:443', text='{"code":403}', status_code=403) + m.get('https://mockKubernetes:443/version', text='{"major": "1.14.10"}', status_code=200) m.get('https://mockOther:443', text='elephant') e = Event() @@ -43,20 +45,20 @@ def test_ApiServerWithServiceAccountToken(): # We should discover an API Server regardless of whether we have a token e.host = 'mockKubernetes' - a = ApiServerDiscovery(e) + a = ApiServiceDiscovery(e) a.execute() time.sleep(0.1) assert counter == 1 e.auth_token = "very_secret" - a = ApiServerDiscovery(e) + a = ApiServiceDiscovery(e) a.execute() time.sleep(0.1) assert counter == 2 - # But we shouldn't generate an event if we don't see an error code + # But we shouldn't generate an event if we don't see an error code or find the 'major' in /version e.host = 'mockOther' - a = ApiServerDiscovery(e) + a = ApiServiceDiscovery(e) a.execute() time.sleep(0.1) assert counter == 2 @@ -78,12 +80,15 @@ def test_InsecureApiServer(): "/apis/apiextensions.k8s.io" ]}""") + m.get('http://mockKubernetes:8080/version', text='{"major": "1.14.10"}') + m.get('http://mockOther:8080/version', status_code=404) + e = Event() e.protocol = "http" e.port = 8080 e.host = 'mockOther' - a = ApiServerDiscovery(e) + a = ApiServiceDiscovery(e) a.execute() e.host = 'mockKubernetes'