1 # -*- coding: utf-8 -*-
4 # Copyright 2020 Telefonica S.A.
6 # Licensed under the Apache License, Version 2.0 (the "License"); you may
7 # not use this file except in compliance with the License. You may obtain
8 # a copy of the License at
10 # http://www.apache.org/licenses/LICENSE-2.0
12 # Unless required by applicable law or agreed to in writing, software
13 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
14 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
15 # License for the specific language governing permissions and limitations
25 from osm_lcm
.lcm_utils
import LcmException
26 from osm_common
.dbbase
import DbException
27 from jinja2
import Template
, TemplateError
, TemplateNotFound
, TemplateSyntaxError
29 __author__
= "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
31 initial_prometheus_data
= {
36 "modified": 1593445184, # 2020-06-29
37 "created": 1593445184,
38 "version": "1.0" # to allow future version updates
40 'scrape_configs': { # Dictionary at database. Converted to list before sending to prometheus
41 'mon_exporter': {'static_configs': [{'targets': ['mon:8000']}], 'job_name': 'mon_exporter'},
43 'global': {'evaluation_interval': '15s', 'scrape_interval': '15s'},
45 'alerting': {'alertmanagers': [{'static_configs': [{'targets': None}]}]}
51 Implements a class to update Prometheus
54 PROMETHEUS_LOCKED_TIME
= 120
56 def __init__(self
, config
, worker_id
, db
, loop
, logger
=None):
57 self
.worker_id
= worker_id
60 self
.logger
= logger
or logging
.getLogger("lcm.prometheus")
61 self
.server
= config
["uri"]
62 self
.path
= config
["path"]
63 if not self
.path
.endswith("/"):
65 self
.cfg_file
= self
.path
+ "prometheus.yml"
66 self
.cfg_file_backup
= self
.path
+ "prometheus.yml-backup"
69 def parse_job(job_data
: str, variables
: dict) -> dict:
71 template
= Template(job_data
)
72 job_parsed
= template
.render(variables
or {})
73 return yaml
.safe_load(job_parsed
)
74 except (TemplateError
, TemplateNotFound
, TemplateSyntaxError
) as e
:
75 # TODO yaml exceptions
76 raise LcmException("Error parsing Jinja2 to prometheus job. job_data={}, variables={}. Error={}".format(
77 job_data
, variables
, e
))
79 async def start(self
):
80 for retry
in range(4):
82 # self.logger("Starting prometheus ")
84 prometheus_data
= self
.db
.get_one("admin", {"_id": "prometheus"}, fail_on_empty
=False)
85 if not prometheus_data
:
86 self
.logger
.info("Init db.admin.prometheus content")
87 self
.db
.create("admin", initial_prometheus_data
)
88 # send database config file to prometheus. Ignore loading errors, as prometheus may be starting
89 # but at least an initial configuration file is set
92 except DbException
as e
:
94 raise LcmException("Max retries trying to init prometheus configuration: {}".format(e
))
95 await asyncio
.sleep(5, loop
=self
.loop
)
97 async def update(self
, add_jobs
: dict = None, remove_jobs
: list = None) -> bool:
100 :param add_jobs: dictionary with {job_id_1: job_content, job_id_2: job_content}
101 :param remove_jobs: list with jobs to remove [job_id_1, job_id_2]
102 :return: result. If false prometheus denies this configuration. Exception on error
104 for retry
in range(20):
106 if retry
: # first time do not wait
107 await asyncio
.sleep(4 + retry
, loop
=self
.loop
)
111 if not self
.db
.set_one(
113 q_filter
={"_id": "prometheus", "_admin.locked_at.lt": now
- self
.PROMETHEUS_LOCKED_TIME
},
114 update_dict
={"_admin.locked_at": now
, "_admin.locked_by": self
.worker_id
},
115 fail_on_empty
=False):
118 prometheus_data
= self
.db
.get_one("admin", {"_id": "prometheus"})
119 update_dict
= {"_admin.locked_at": 0,
120 "_admin.locked_by": None}
122 # Make changes from prometheus_incremental
123 push_dict
= pull_dict
= None
124 if add_jobs
or remove_jobs
:
127 log_text_list
.append("adding jobs: {}".format(list(add_jobs
.keys())))
128 prometheus_data
["scrape_configs"].update(add_jobs
)
129 push_dict
= {"scrape_configs." + job_id
: job_data
for job_id
, job_data
in add_jobs
.items()}
131 log_text_list
.append("removing jobs: {}".format(list(remove_jobs
)))
132 for job_id
in remove_jobs
:
133 prometheus_data
["scrape_configs"].pop(job_id
, None)
134 pull_dict
= {"scrape_configs." + job_id
: None for job_id
in remove_jobs
}
135 self
.logger
.debug("Updating. " + ". ".join(log_text_list
))
137 if not await self
.send_data(prometheus_data
):
138 self
.logger
.error("Cannot update add_jobs: {}. remove_jobs: {}".format(add_jobs
, remove_jobs
))
139 push_dict
= pull_dict
= None
144 update_dict
.update(push_dict
)
145 if push_dict
or pull_dict
:
146 update_dict
["_admin.modified_at"] = now
147 if not self
.db
.set_one(
148 "admin", {"_id": "prometheus", "_admin.locked_at": now
, "_admin.locked_by": self
.worker_id
},
149 update_dict
=update_dict
, unset
=pull_dict
, fail_on_empty
=False):
152 raise LcmException("Cannot update prometheus database. Reached max retries")
154 async def send_data(self
, new_config
):
155 restore_backup
= False
156 del new_config
["_id"]
157 del new_config
["_admin"]
158 new_scrape_configs
= []
160 # generate a list with the values of scrape_configs
161 for scrape_config
in new_config
["scrape_configs"].values():
162 scrape_config
= scrape_config
.copy()
163 # remove nsr_id metadata from scrape_configs
164 scrape_config
.pop("nsr_id", None)
165 new_scrape_configs
.append(scrape_config
)
166 new_config
["scrape_configs"] = new_scrape_configs
169 if os
.path
.exists(self
.cfg_file
):
170 os
.rename(self
.cfg_file
, self
.cfg_file_backup
)
171 restore_backup
= True
172 with
open(self
.cfg_file
, "w+") as f
:
173 yaml
.safe_dump(new_config
, f
, indent
=4, default_flow_style
=False)
174 # self.logger.debug("new configuration: {}".format(yaml.safe_dump(new_config, indent=4,
175 # default_flow_style=False)))
176 async with aiohttp
.ClientSession() as session
:
177 async with session
.post(self
.server
+ "-/reload") as resp
:
178 if resp
.status
> 204:
179 raise LcmException(await resp
.text())
180 await asyncio
.sleep(5, loop
=self
.loop
)
181 # If prometheus does not admit this configuration, remains with the old one
182 # Then, to check if the configuration has been accepted, get the configuration from prometheus
183 # and compares with the inserted one
184 async with session
.get(self
.server
+ "api/v1/status/config") as resp
:
185 if resp
.status
> 204:
186 raise LcmException(await resp
.text())
187 current_config
= await resp
.json()
188 if not self
._check
_configuration
_equal
(current_config
, new_config
):
191 restore_backup
= False
193 except Exception as e
:
194 self
.logger
.error("Error updating configuration url={}: {}".format(self
.server
, e
))
199 os
.rename(self
.cfg_file_backup
, self
.cfg_file
)
200 except Exception as e
:
201 self
.logger
.critical("Exception while rolling back: {}".format(e
))
203 def _check_configuration_equal(self
, current_config
, expected_config
):
205 # self.logger.debug("Comparing current_config='{}' with expected_config='{}'".format(current_config,
207 current_config_yaml
= yaml
.safe_load(current_config
['data']['yaml'])
208 current_jobs
= [j
["job_name"] for j
in current_config_yaml
["scrape_configs"]]
209 expected_jobs
= [j
["job_name"] for j
in expected_config
["scrape_configs"]]
210 if current_jobs
== expected_jobs
:
213 self
.logger
.error("Not all jobs have been loaded. Target jobs: {} Loaded jobs: {}".format(
214 expected_jobs
, current_jobs
))
216 except Exception as e
:
217 self
.logger
.error("Invalid obtained status from server. Error: '{}'. Obtained data: '{}'".format(
219 # if format is not understood, cannot be compared, assume it is ok