ceph/src/pybind/mgr/cephadm/services/jaeger.py

   1 from typing import List, cast
   2 from cephadm.services.cephadmservice import CephadmService, CephadmDaemonDeploySpec
   3 from ceph.deployment.service_spec import TracingSpec
   4 from mgr_util import build_url
   5
   6
   7 class ElasticSearchService(CephadmService):
   8     TYPE = 'elasticsearch'
   9     DEFAULT_SERVICE_PORT = 9200
  10
  11     def prepare_create(self, daemon_spec: CephadmDaemonDeploySpec) -> CephadmDaemonDeploySpec:
  12         assert self.TYPE == daemon_spec.daemon_type
  13         return daemon_spec
  14
  15
  16 class JaegerAgentService(CephadmService):
  17     TYPE = 'jaeger-agent'
  18     DEFAULT_SERVICE_PORT = 6799
  19
  20     def prepare_create(self, daemon_spec: CephadmDaemonDeploySpec) -> CephadmDaemonDeploySpec:
  21         assert self.TYPE == daemon_spec.daemon_type
  22         collectors = []
  23         for dd in self.mgr.cache.get_daemons_by_type(JaegerCollectorService.TYPE):
  24             # scrape jaeger-collector nodes
  25             assert dd.hostname is not None
  26             port = dd.ports[0] if dd.ports else JaegerCollectorService.DEFAULT_SERVICE_PORT
  27             url = build_url(host=dd.hostname, port=port).lstrip('/')
  28             collectors.append(url)
  29         daemon_spec.final_config = {'collector_nodes': ",".join(collectors)}
  30         return daemon_spec
  31
  32
  33 class JaegerCollectorService(CephadmService):
  34     TYPE = 'jaeger-collector'
  35     DEFAULT_SERVICE_PORT = 14250
  36
  37     def prepare_create(self, daemon_spec: CephadmDaemonDeploySpec) -> CephadmDaemonDeploySpec:
  38         assert self.TYPE == daemon_spec.daemon_type
  39         elasticsearch_nodes = get_elasticsearch_nodes(self, daemon_spec)
  40         daemon_spec.final_config = {'elasticsearch_nodes': ",".join(elasticsearch_nodes)}
  41         return daemon_spec
  42
  43
  44 class JaegerQueryService(CephadmService):
  45     TYPE = 'jaeger-query'
  46     DEFAULT_SERVICE_PORT = 16686
  47
  48     def prepare_create(self, daemon_spec: CephadmDaemonDeploySpec) -> CephadmDaemonDeploySpec:
  49         assert self.TYPE == daemon_spec.daemon_type
  50         elasticsearch_nodes = get_elasticsearch_nodes(self, daemon_spec)
  51         daemon_spec.final_config = {'elasticsearch_nodes': ",".join(elasticsearch_nodes)}
  52         return daemon_spec
  53
  54
  55 def get_elasticsearch_nodes(service: CephadmService, daemon_spec: CephadmDaemonDeploySpec) -> List[str]:
  56     elasticsearch_nodes = []
  57     for dd in service.mgr.cache.get_daemons_by_type(ElasticSearchService.TYPE):
  58         assert dd.hostname is not None
  59         addr = dd.ip if dd.ip else service.mgr.inventory.get_addr(dd.hostname)
  60         port = dd.ports[0] if dd.ports else ElasticSearchService.DEFAULT_SERVICE_PORT
  61         url = build_url(host=addr, port=port).lstrip('/')
  62         elasticsearch_nodes.append(f'http://{url}')
  63
  64     if len(elasticsearch_nodes) == 0:
  65         # takes elasticsearch address from TracingSpec
  66         spec: TracingSpec = cast(
  67             TracingSpec, service.mgr.spec_store.active_specs[daemon_spec.service_name])
  68         assert spec.es_nodes is not None
  69         urls = spec.es_nodes.split(",")
  70         for url in urls:
  71             elasticsearch_nodes.append(f'http://{url}')
  72
  73     return elasticsearch_nodes