I run into a small stumbling block the other evening while working on my ‘site domain manager’ project (for want of a better name). This is essentially a REST API running in a daemon service that manages the mappings of domains to websites, and uses ‘agents’ to automate the configuration via API calls to the various services involved (domain registrars, DNS servers, WAF providers, SSL certs etc.)
The problem arose because the manager class I was writing to interact with kubernetes needed to manage certificates. The kubernetes python client library has a whole bunch of useful higher-level API and model classes for listing, creating, updating the main models I needed to manage, such as the ConfigMap, DaemonSet, Service and Ingress, but because the Certificate is part of the cert-manager package, it doesn’t have the equivalent higher-level methods I needed.
In the end, we solved it by using the lower-level call_api method, as follows:
siteid = sitename.split(".")[0]
rawyaml = """
apiVersion: certmanager.k8s.io/v1alpha1
kind: Certificate
metadata:
name: {{siteid}}-cert
namespace: {{namespace}}
spec:
acme:
config:
- dns01:
provider: route53
domains:
- {{mainhostname}}
commonName: {{mainhostname}}
dnsNames:
- {{mainhostname}}
issuerRef:
kind: ClusterIssuer
name: letsencrypt-production
secretName: {{siteid}}-tls
"""
rawyaml = rawyaml.replace("{{siteid}}", siteid)
rawyaml = rawyaml.replace("{{namespace}}", self.namespace)
rawyaml = rawyaml.replace("{{mainhostname}}", mainhostname)
model = yaml.load(rawyaml, Loader=yaml.SafeLoader)
model['spec']['acme']['config'][0]['domains'] = aliases
model['spec']['dnsNames'] = aliases
path_params = {}
auth_settings = ['BearerToken']
header_params = {
"Content-Type": "application/json"
}
query_params = []
# Fetch latest state of resource to apply changes to
response = self.api_client.call_api(f"/apis/certmanager.k8s.io/v1alpha1/namespaces/{self.namespace}/certificates", 'GET',
path_params,
query_params,
header_params,
auth_settings=auth_settings,
)
r = json.loads(self.api_client.last_response.data)
certs = {x['metadata']['name']:x for x in r['items']}
# If it doesn't exist, create it...
certid = f"{siteid}-cert"
try:
if certid not in certs.keys():
_logger.info(f"Creating certificate '{certid}' with {len(aliases)} hostnames...")
response = self.api_client.call_api(f"/apis/certmanager.k8s.io/v1alpha1/namespaces/{self.namespace}/certificates", 'POST',
path_params,
query_params,
header_params,
body=model,
auth_settings=auth_settings,
)
else:
_logger.info(f"Updating certificate '{certid}' with {len(aliases)} hostnames...")
# Transplant metadata to allow update to work
model['metadata'] = certs[certid]['metadata']
# Attempt to update...
response = self.api_client.call_api(f"/apis/certmanager.k8s.io/v1alpha1/namespaces/{self.namespace}/certificates/{certid}", 'PUT',
path_params,
query_params,
header_params,
body=model,
auth_settings=auth_settings,
)
except Exception as e:
_logger.exception(e)
return str(e)
You can see this in context here .