149 lines
4.1 KiB
Python
149 lines
4.1 KiB
Python
# SPDX-FileCopyrightText: 2023 Gregor Kleen
|
||
#
|
||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||
|
||
import os, sys
|
||
from os import environ
|
||
|
||
import kubernetes as k8s
|
||
|
||
from pathlib import Path
|
||
|
||
import argparse
|
||
import logging
|
||
|
||
from base64 import b64decode
|
||
|
||
from minio import Minio
|
||
|
||
|
||
def main():
|
||
global logger
|
||
logger = logging.getLogger(__name__)
|
||
console_handler = logging.StreamHandler()
|
||
console_handler.setFormatter(
|
||
logging.Formatter("[%(levelname)s](%(name)s): %(message)s")
|
||
)
|
||
if sys.stderr.isatty():
|
||
console_handler.setFormatter(
|
||
logging.Formatter("%(asctime)s [%(levelname)s](%(name)s): %(message)s")
|
||
)
|
||
logger.addHandler(console_handler)
|
||
|
||
# log uncaught exceptions
|
||
def log_exceptions(type, value, tb):
|
||
global logger
|
||
|
||
logger.error(value)
|
||
sys.__excepthook__(type, value, tb) # calls default excepthook
|
||
|
||
sys.excepthook = log_exceptions
|
||
|
||
parser = argparse.ArgumentParser(
|
||
prog="k8s-gitlab-borg", formatter_class=argparse.ArgumentDefaultsHelpFormatter
|
||
)
|
||
parser.add_argument(
|
||
"--verbosity",
|
||
dest="log_level",
|
||
action="append",
|
||
type=int,
|
||
help="Set verbosity to specific level",
|
||
)
|
||
parser.add_argument(
|
||
"--verbose",
|
||
"-v",
|
||
dest="log_level",
|
||
action="append_const",
|
||
const=1,
|
||
help="Increase verbosity by one step",
|
||
)
|
||
parser.add_argument(
|
||
"--quiet",
|
||
"-q",
|
||
dest="log_level",
|
||
action="append_const",
|
||
const=-1,
|
||
help="Decrease verbosity by one step",
|
||
)
|
||
parser.add_argument(
|
||
"--k8s-config",
|
||
type=Path,
|
||
metavar="FILE",
|
||
default=Path(environ["CREDENTIALS_DIRECTORY"]) / "k8s.yaml"
|
||
if "CREDENTIALS_DIRECTORY" in environ
|
||
else k8s.config.KUBE_CONFIG_DEFAULT_LOCATION,
|
||
help="Path to YAML file encoding k8s credentials",
|
||
)
|
||
parser.add_argument(
|
||
"--namespace",
|
||
type=str,
|
||
metavar="STR",
|
||
default="gitlab",
|
||
help="K8s namespace containing gitlab deployment(s)",
|
||
)
|
||
args = parser.parse_args()
|
||
|
||
LOG_LEVELS = [
|
||
logging.DEBUG,
|
||
logging.INFO,
|
||
logging.WARNING,
|
||
logging.ERROR,
|
||
logging.CRITICAL,
|
||
]
|
||
DEFAULT_LOG_LEVEL = logging.ERROR
|
||
log_level = LOG_LEVELS.index(DEFAULT_LOG_LEVEL)
|
||
|
||
for adjustment in args.log_level or ():
|
||
log_level = min(len(LOG_LEVELS) - 1, max(log_level - adjustment, 0))
|
||
logger.setLevel(LOG_LEVELS[log_level])
|
||
|
||
logger.debug("K8s config file: ‘%s’", args.k8s_config)
|
||
with k8s.config.new_client_from_config(str(args.k8s_config)) as client:
|
||
core_api = k8s.client.CoreV1Api(client)
|
||
|
||
services = core_api.list_namespaced_service(namespace=args.namespace).items
|
||
|
||
def is_minio(item):
|
||
selector = item.spec.selector
|
||
|
||
if "app" not in selector:
|
||
return False
|
||
if selector["app"] != "minio":
|
||
return False
|
||
if "release" not in selector:
|
||
return False
|
||
if selector["release"] != "gitlab":
|
||
return False
|
||
|
||
return True
|
||
|
||
[minio_svc] = list(filter(is_minio, services))
|
||
|
||
secrets = core_api.list_namespaced_secret(namespace=args.namespace).items
|
||
|
||
def is_minio(item):
|
||
return item.metadata.name == "gitlab-minio-secret"
|
||
|
||
[minio_secret] = list(filter(is_minio, secrets))
|
||
|
||
minio_host = minio_svc.spec.cluster_ip
|
||
[minio_port] = minio_svc.spec.ports
|
||
minio_port = minio_port.port
|
||
logger.debug("Determined minio host: %s:%d", minio_host, minio_port)
|
||
minio_accesskey = b64decode(minio_secret.data["accesskey"]).decode("utf-8")
|
||
minio_secretkey = b64decode(minio_secret.data["secretkey"]).decode("utf-8")
|
||
|
||
minio = Minio(
|
||
f"{minio_host}:{minio_port:d}",
|
||
secure=False,
|
||
access_key=minio_accesskey,
|
||
secret_key=minio_secretkey,
|
||
)
|
||
|
||
for obj in minio.list_objects("gitlab-backups"):
|
||
logger.debug("‘%s’", obj.object_name)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|