Skip to content

Commit

Permalink
feat: add ability to fetch only specific resources by name
Browse files Browse the repository at this point in the history
  • Loading branch information
@simonsystem
simonsystem committed Dec 19, 2024
1 parent b847cd3 commit e171026
Show file tree
Hide file tree
Showing 3 changed files with 56 additions and 18 deletions.
1 change: 1 addition & 0 deletions README.md
View file
Original file line number Diff line number Diff line change
Expand Up @@ -71,6 +71,7 @@ If the filename ends with `.url` suffix, the content will be processed as a URL
| `FOLDER_ANNOTATION` | The annotation the sidecar will look for in configmaps to override the destination folder for files. The annotation _value_ can be either an absolute or a relative path. Relative paths will be relative to `FOLDER`. | false | `k8s-sidecar-target-directory` | string |
| `NAMESPACE` | Comma separated list of namespaces. If specified, the sidecar will search for config-maps inside these namespaces. It's also possible to specify `ALL` to search in all namespaces. | false | namespace in which the sidecar is running | string |
| `RESOURCE` | Resource type, which is monitored by the sidecar. Options: `configmap`, `secret`, `both` | false | `configmap` | string |
| `RESOURCE_NAME` | Comma separated list of resource names, which are monitored by the sidecar. Items can be prefixed by the namespace and the resource type. E.g. `namespace/resource-name` or `secret/namespace/resource-name`. Setting this will result `method` set to `WATCH` being treated as `SLEEP` | false | - | string |
| `METHOD` | If `METHOD` is set to `LIST`, the sidecar will just list config-maps/secrets and exit. With `SLEEP` it will list all config-maps/secrets, then sleep for `SLEEP_TIME` seconds. Anything else will continuously watch for changes (see https://kubernetes.io/docs/reference/using-api/api-concepts/#efficient-detection-of-changes). | false | - | string |
| `SLEEP_TIME` | How many seconds to wait before updating config-maps/secrets when using `SLEEP` method. | false | `60` | integer |
| `REQ_URL` | URL to which send a request after a configmap/secret got reloaded | false | - | URI |
Expand Down
65 changes: 49 additions & 16 deletions src/resources.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,11 @@
RESOURCE_CONFIGMAP: "list_config_map_for_all_namespaces"
}})

_read_namespace = {
RESOURCE_SECRET: "read_namespaced_secret",
RESOURCE_CONFIGMAP: "read_namespaced_config_map"
}

_resources_version_map = {
RESOURCE_SECRET: {},
RESOURCE_CONFIGMAP: {},
Expand Down Expand Up @@ -98,26 +103,48 @@ def _get_destination_folder(metadata, default_folder, folder_annotation):

def list_resources(label, label_value, target_folder, request_url, request_method, request_payload,
namespace, folder_annotation, resource, unique_filenames, script, enable_5xx,
ignore_already_processed):
ignore_already_processed, resource_name):
v1 = client.CoreV1Api()
# Filter resources based on label and value or just label
label_selector = f"{label}={label_value}" if label_value else label

additional_args = {
'label_selector': label_selector
}
additional_args = {}

if namespace != "ALL":
additional_args['namespace'] = namespace

logger.info(f"Performing list-based sync on {resource} resources: {additional_args}")

ret = getattr(v1, _list_namespace[namespace][resource])(**additional_args)
resource_names = []

if namespace != "ALL" and resource_name:
for rn in resource_name.split(","):
splitted_rn = list(reversed(rn.split("/")))
if len(splitted_rn) == 3 and splitted_rn[2] != resource:
continue
if len(splitted_rn) == 2 and splitted_rn[1] != namespace:
continue
resource_names.append(splitted_rn[0])

if namespace != "ALL" and resource_names:
items = []
for rn in resource_names:
additional_args['name'] = rn
try:
ret = getattr(v1, _read_namespace[resource])(**additional_args)
items.append(ret)
except ApiException as e:
if e.status != 404:
raise e

else:
additional_args['label_selector'] = f"{label}={label_value}" if label_value else label
ret = getattr(v1, _list_namespace[namespace][resource])(**additional_args)
items = ret.items

files_changed = False
exist_keys = set()

# For all the found resources
for item in ret.items:
for item in items:
metadata = item.metadata
exist_keys.add(metadata.namespace + metadata.name)

Expand Down Expand Up @@ -362,14 +389,20 @@ def _watch_resource_iterator(label, label_value, target_folder, request_url, req
request(request_url, request_method, enable_5xx, request_payload)


def _watch_resource_loop(mode, *args):
def _watch_resource_loop(mode, label, label_value, target_folder, request_url, request_method, request_payload,
namespace, folder_annotation, resource, unique_filenames, script, enable_5xx,
ignore_already_processed, resource_name):
while True:
try:
if mode == "SLEEP":
list_resources(*args)
if mode == "SLEEP" or (namespace != 'ALL' and resource_name):
list_resources(label, label_value, target_folder, request_url, request_method, request_payload,
namespace, folder_annotation, resource, unique_filenames, script, enable_5xx,
ignore_already_processed, resource_name)
sleep(int(os.getenv("SLEEP_TIME", 60)))
else:
_watch_resource_iterator(*args)
_watch_resource_iterator(label, label_value, target_folder, request_url, request_method, request_payload,
namespace, folder_annotation, resource, unique_filenames, script, enable_5xx,
ignore_already_processed)
except ApiException as e:
if e.status != 500:
logger.error(f"ApiException when calling kubernetes: {e}\n")
Expand All @@ -389,11 +422,11 @@ def _watch_resource_loop(mode, *args):

def watch_for_changes(mode, label, label_value, target_folder, request_url, request_method, request_payload,
current_namespace, folder_annotation, resources, unique_filenames, script, enable_5xx,
ignore_already_processed):
ignore_already_processed, resource_name):
processes = _start_watcher_processes(current_namespace, folder_annotation, label,
label_value, request_method, mode, request_payload, resources,
target_folder, unique_filenames, script, request_url, enable_5xx,
ignore_already_processed)
ignore_already_processed, resource_name)

while True:
died = False
Expand All @@ -413,14 +446,14 @@ def watch_for_changes(mode, label, label_value, target_folder, request_url, requ

def _start_watcher_processes(namespace, folder_annotation, label, label_value, request_method,
mode, request_payload, resources, target_folder, unique_filenames, script, request_url,
enable_5xx, ignore_already_processed):
enable_5xx, ignore_already_processed, resource_name):
processes = []
for resource in resources:
for ns in namespace.split(','):
proc = Process(target=_watch_resource_loop,
args=(mode, label, label_value, target_folder, request_url, request_method, request_payload,
ns, folder_annotation, resource, unique_filenames, script, enable_5xx,
ignore_already_processed)
ignore_already_processed, resource_name)
)
proc.daemon = True
proc.start()
Expand Down
8 changes: 6 additions & 2 deletions src/sidecar.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
LABEL = "LABEL"
LABEL_VALUE = "LABEL_VALUE"
RESOURCE = "RESOURCE"
RESOURCE_NAME = "RESOURCE_NAME"
REQ_PAYLOAD = "REQ_PAYLOAD"
REQ_URL = "REQ_URL"
REQ_METHOD = "REQ_METHOD"
Expand Down Expand Up @@ -70,6 +71,9 @@ def main():
resources = ("secret", "configmap") if resources == "both" else (resources,)
logger.debug(f"Selected resource type: {resources}")

resource_name = os.getenv(RESOURCE_NAME, "")
logger.debug(f"Selected resource name: {resource_name}")

request_method = os.getenv(REQ_METHOD)
request_url = os.getenv(REQ_URL)

Expand Down Expand Up @@ -127,11 +131,11 @@ def main():
for ns in namespace.split(','):
list_resources(label, label_value, target_folder, request_url, request_method, request_payload,
ns, folder_annotation, res, unique_filenames, script, enable_5xx,
ignore_already_processed)
ignore_already_processed, resource_name)
else:
watch_for_changes(method, label, label_value, target_folder, request_url, request_method, request_payload,
namespace, folder_annotation, resources, unique_filenames, script, enable_5xx,
ignore_already_processed)
ignore_already_processed, resource_name)


def _initialize_kubeclient_configuration():
Expand Down

0 comments on commit e171026

Please sign in to comment.