from urllib.error import HTTPError
from threading import Event
-from cephadm.service_discovery import ServiceDiscovery
-
from ceph.deployment.service_spec import PrometheusSpec
import string
@handle_orch_error
def service_discovery_dump_cert(self) -> str:
- root_cert = self.get_store(ServiceDiscovery.KV_STORE_SD_ROOT_CERT)
+ root_cert = self.cert_key_store.get_cert('service_discovery_root_cert')
if not root_cert:
raise OrchestratorError('No certificate found for service discovery')
return root_cert
class ServiceDiscovery:
+ # TODO: these constants should only be needed for migration purposes
+ # after completion of the cert store. Make sure to move them.
KV_STORE_SD_ROOT_CERT = 'service_discovery/root/cert'
KV_STORE_SD_ROOT_KEY = 'service_discovery/root/key'
self.mgr.set_store('service_discovery/root/username', self.username)
def configure_tls(self, server: Server) -> None:
- old_cert = self.mgr.get_store(self.KV_STORE_SD_ROOT_CERT)
- old_key = self.mgr.get_store(self.KV_STORE_SD_ROOT_KEY)
+ old_cert = self.mgr.cert_key_store.get_cert('service_discovery_root_cert')
+ old_key = self.mgr.cert_key_store.get_key('service_discovery_key')
if old_key and old_cert:
self.ssl_certs.load_root_credentials(old_cert, old_key)
else:
self.ssl_certs.generate_root_cert(self.mgr.get_mgr_ip())
- self.mgr.set_store(self.KV_STORE_SD_ROOT_CERT, self.ssl_certs.get_root_cert())
- self.mgr.set_store(self.KV_STORE_SD_ROOT_KEY, self.ssl_certs.get_root_key())
+ self.mgr.cert_key_store.save_cert('service_discovery_root_cert', self.ssl_certs.get_root_cert())
+ self.mgr.cert_key_store.save_key('service_discovery_key', self.ssl_certs.get_root_key())
addr = self.mgr.get_mgr_ip()
host_fqdn = socket.getfqdn(addr)
server.ssl_certificate, server.ssl_private_key = self.ssl_certs.generate_cert_files(