This repository has been archived on 2020-03-24. You can view files and clone it, but cannot push or open issues or pull requests.
quay/config_app/config_util/config/TransientDirectoryProvider.py

63 lines
2.1 KiB
Python
Raw Normal View History

import os
import base64
from backports.tempfile import TemporaryDirectory
from config_app.config_util.config.fileprovider import FileConfigProvider
from config_app.config_util.k8saccessor import KubernetesAccessorSingleton
from util.config.validator import EXTRA_CA_DIRECTORY, EXTRA_CA_DIRECTORY_PREFIX
class TransientDirectoryProvider(FileConfigProvider):
""" Implementation of the config provider that reads and writes the data
from/to the file system, only using temporary directories,
deleting old dirs and creating new ones as requested.
"""
def __init__(self, config_volume, yaml_filename, py_filename):
# Create a temp directory that will be cleaned up when we change the config path
# This should ensure we have no "pollution" of different configs:
# no uploaded config should ever affect subsequent config modifications/creations
temp_dir = TemporaryDirectory()
self.temp_dir = temp_dir
super(TransientDirectoryProvider, self).__init__(temp_dir.name, yaml_filename, py_filename)
@property
def provider_id(self):
return 'transient'
def new_config_dir(self):
"""
Update the path with a new temporary directory, deleting the old one in the process
"""
self.temp_dir.cleanup()
temp_dir = TemporaryDirectory()
self.config_volume = temp_dir.name
self.temp_dir = temp_dir
self.yaml_path = os.path.join(temp_dir.name, self.yaml_filename)
def get_config_dir_path(self):
return self.config_volume
def save_configuration_to_kubernetes(self):
data = {}
certs_dir = os.path.join(self.config_volume, EXTRA_CA_DIRECTORY)
if os.path.exists(certs_dir):
for extra_cert in os.listdir(certs_dir):
with open(os.path.join(certs_dir, extra_cert)) as f:
data[EXTRA_CA_DIRECTORY_PREFIX + extra_cert] = base64.b64encode(f.read())
for name in os.listdir(self.config_volume):
file_path = os.path.join(self.config_volume, name)
if not os.path.isdir(file_path):
with open(file_path) as f:
data[name] = base64.b64encode(f.read())
KubernetesAccessorSingleton.get_instance().replace_qe_secret(data)
return 200