Coverage for python/lsst/analysis/tools/interfaces/datastore/_dispatcher.py: 24%
305 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-17 10:03 +0000
1# This file is part of analysis_tools.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (http://www.lsst.org).
6# See the COPYRIGHT file at the top-level directory of this distribution
7# for details of code ownership.
8#
9# This program is free software: you can redistribute it and/or modify
10# it under the terms of the GNU General Public License as published by
11# the Free Software Foundation, either version 3 of the License, or
12# (at your option) any later version.
13#
14# This program is distributed in the hope that it will be useful,
15# but WITHOUT ANY WARRANTY; without even the implied warranty of
16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
17# GNU General Public License for more details.
18#
19# You should have received a copy of the GNU General Public License
20# along with this program. If not, see <http://www.gnu.org/licenses/>.
22from __future__ import annotations
24__all__ = ("SasquatchDispatchPartialFailure", "SasquatchDispatchFailure", "SasquatchDispatcher")
26import calendar
27import datetime
28import json
29import logging
30import math
31import re
32from collections.abc import Mapping, MutableMapping, Sequence
33from dataclasses import dataclass
34from typing import TYPE_CHECKING, Any, cast
35from uuid import UUID, uuid4
37import numpy as np
38import requests
40from lsst.daf.butler import DatasetRef
41from lsst.resources import ResourcePath
42from lsst.utils.packages import getEnvironmentPackages
44from ...utils import http_client
46if TYPE_CHECKING:
47 from .. import MetricMeasurementBundle
49"""Sasquatch datastore"""
51log = logging.getLogger(__name__)
53# Constants associated with SasquatchDispatcher
54PARTITIONS = 1
55REPLICATION_FACTOR = 3
57DEFAULT_TIMEOUT: float = 30.0
58"""Default timeout in seconds applied to each HTTP request to the proxy."""
60IDENTIFIER_KEYS = [
61 "detector",
62 "patch",
63 "skymap",
64 "visit",
65 "tract",
66 "physical_filter",
67 "instrument",
68 "band",
69 "exposure",
70 "group",
71 "day_obs",
72]
75class SasquatchDispatchPartialFailure(RuntimeError):
76 """This indicates that a Sasquatch dispatch was partially successful."""
78 pass
81class SasquatchDispatchFailure(RuntimeError):
82 """This indicates that dispatching a
83 `~lsst.analysis.tool.interface.MetricMeasurementBundle` failed.
84 """
86 pass
89def _tag2VersionTime(productStr: str) -> tuple[str, float]:
90 """Determine versions and dates from the string returned from
91 getEnvironmentPackages.
93 The `~lsst.utils.packages.genEnvironmentPackages` function returns the
94 setup version associated with a product, along with a list of tags that
95 have been added to it.
97 This method splits up that return string, and determines the earliest date
98 associated with the setup package version.
100 Parameters
101 ----------
102 productStr : `str`
103 The product string returned from a lookup on the result of a call to
104 `~lsst.utils.packages.getEnvironmentPackages`.
106 Returns
107 -------
108 result : `tuple` of `str`, `datetime.datetime`
109 The first `str` is the version of the package, and the second is the
110 datetime object associated with that released version.
112 Raises
113 ------
114 ValueError
115 Raised if there are no tags which correspond to dates.
116 """
117 times: list[datetime.datetime] = []
118 version = productStr.split()[0]
119 pattern_match = re.findall("[(](.*)[)]", productStr)
120 if len(pattern_match) == 0:
121 raise ValueError(f"Could not find any tags in product string {productStr}")
122 tags: str = pattern_match[0]
123 for tag in tags.split():
124 numDots = tag.count(".")
125 numUnder = tag.count("_")
126 separator = "_"
127 if numDots > numUnder:
128 separator = "."
129 match tag.split(separator):
130 # Daily tag branch.
131 case ("d", year, month, day):
132 dt = datetime.datetime(year=int(year), month=int(month), day=int(day))
133 # Weekly tag branch.
134 case ("w", year, week):
135 iyear = int(year)
136 iweek = int(week)
137 # Use 4 as the day because releases are available starting
138 # on Thursday
139 dayOfWeek = 4
141 # Find the first week to contain a thursday in it
142 cal = calendar.Calendar()
143 cal.setfirstweekday(6)
144 i = 0
145 for i, iterWeek in enumerate(cal.monthdatescalendar(iyear, 1)):
146 if iterWeek[dayOfWeek].month == 1:
147 break
148 # Handle fromisocalendar not being able to handle week 53
149 # in the case were the date was going to subtract 7 days anyway
150 if i and iweek == 53:
151 i = 0
152 iweek = 52
153 delta = datetime.timedelta(days=7 * i)
155 # Correct for a weekly being issued in the last week of the
156 # previous year, as Thursdays don't always line up evenly in
157 # a week / year split.
158 dt = datetime.datetime.fromisocalendar(iyear, iweek, dayOfWeek) - delta
159 # Skip tags that can't be understood.
160 case _:
161 continue
162 times.append(dt)
163 if len(times) == 0:
164 raise ValueError("Could not find any tags corresponding to dates")
165 minTime = min(times)
166 minTime.replace(tzinfo=datetime.UTC)
167 return version, minTime.timestamp()
170@dataclass
171class SasquatchDispatcher:
172 """This class mediates the transfer of MetricMeasurementBundles to a
173 Sasquatch http kafka proxy server.
174 """
176 url: str
177 """Url of the Sasquatch proxy server"""
179 token: str
180 """Authentication token used in communicating with the proxy server"""
182 namespace: str = "lsst.dm"
183 """The namespace in Sasquatch in which to write the uploaded metrics"""
185 timeout: float | None = DEFAULT_TIMEOUT
186 """Timeout in seconds for each HTTP request to the proxy, or `None` for
187 no timeout."""
189 def __post_init__(self) -> None:
190 match ResourcePath(self.url).scheme:
191 case "http" | "https": 191 ↛ 193line 191 didn't jump to line 193 because the pattern on line 191 always matched
192 pass
193 case _:
194 raise ValueError("Proxy server must be locatable with either http or https")
196 self._cluster_id: str | None = None
198 @property
199 def clusterId(self) -> str:
200 """ClusterId of the Kafka proxy
202 Notes
203 -----
204 The cluster Id will be fetched with a network call if it is not
205 already cached.
206 """
207 if self._cluster_id is None:
208 self._populateClusterId()
209 return cast(str, self._cluster_id)
211 def _populateClusterId(self) -> None:
212 """Get Sasquatch kafka cluster ID."""
214 headers = {"content-type": "application/json", "Authorization": f"Bearer {self.token}"}
216 try:
217 with http_client() as session:
218 r = session.get(f"{self.url}/v3/clusters", headers=headers, timeout=self.timeout)
219 r.raise_for_status()
220 cluster_id = r.json()["data"][0]["cluster_id"]
221 self._cluster_id = str(cluster_id)
222 except requests.RequestException as e:
223 raise SasquatchDispatchFailure("Could not retrieve the cluster id for the specified url") from e
225 def _create_topic(self, topic_name: str) -> bool:
226 """Create a kafka topic in Sasquatch.
228 Parameters
229 ----------
230 topic_name : `str`
231 The name of the kafka topic to create
233 Returns
234 -------
235 status : `bool`
236 If this does not encounter an error it will return a True success
237 code, else it will return a False code.
239 """
241 headers = {"content-type": "application/json", "Authorization": f"Bearer {self.token}"}
243 topic_config = {
244 "topic_name": f"{self.namespace}.{topic_name}",
245 "partitions_count": PARTITIONS,
246 "replication_factor": REPLICATION_FACTOR,
247 }
249 try:
250 with http_client() as session:
251 r = session.post(
252 f"{self.url}/v3/clusters/{self.clusterId}/topics",
253 json=topic_config,
254 headers=headers,
255 timeout=self.timeout,
256 )
257 r.raise_for_status()
258 log.debug("Created topic %s.%s", self.namespace, topic_name)
259 return True
260 except requests.HTTPError as e:
261 if e.response.status_code == requests.codes.bad_request:
262 log.debug("Topic %s.%s already exists.", self.namespace, topic_name)
263 return True
264 else:
265 log.error(
266 "Unknown error occurred creating kafka topic %s %s",
267 e.response.status_code,
268 e.response.reason,
269 )
270 return False
271 except requests.RequestException:
272 log.exception("Unknown error occurred creating kafka topic")
273 return False
275 def _generateAvroSchema(self, metric: str, record: MutableMapping[str, Any]) -> tuple[str, bool]:
276 """Infer the Avro schema from the record payload.
278 Parameters
279 ----------
280 metric : `str`
281 The name of the metric
282 record : `MutableMapping`
283 The prepared record for which a schema is to be generated
285 Returns
286 -------
287 resultSchema : `str`
288 A json encoded string of the resulting avro schema
289 errorCode : bool
290 A boolean indicating if any record fields had to be trimmed because
291 a suitable schema could not be generated. True if records were
292 removed, False otherwise.
293 """
294 schema: dict[str, Any] = {"type": "record", "namespace": self.namespace, "name": metric}
296 # Record if any records needed to be trimmed
297 resultsTrimmed = False
299 fields = list()
300 # If avro schemas cant be generated for values, they should be removed
301 # from the records.
302 keysToRemove: list[str] = []
303 for key in record:
304 value = record[key]
305 avroType: Mapping[str, Any]
306 if "timestamp" in key:
307 avroType = {"type": "double"}
308 else:
309 avroType = self._python2Avro(value)
310 if len(avroType) == 0:
311 continue
312 if avroType.get("error_in_conversion"):
313 keysToRemove.append(key)
314 resultsTrimmed = True
315 continue
316 fields.append({"name": key, **avroType})
318 # remove any key that failed to have schema generated
319 for key in keysToRemove:
320 record.pop(key)
322 schema["fields"] = fields
324 return json.dumps(schema), resultsTrimmed
326 def _python2Avro(self, value: Any) -> Mapping:
327 """Map python type to avro schema
329 Parameters
330 ----------
331 value : `Any`
332 Any python parameter.
334 Returns
335 -------
336 result : `Mapping`
337 Return a mapping that represents an entry in an avro schema.
338 """
339 match value:
340 case float() | np.float32() | None:
341 return {"type": "float", "default": 0.0}
342 case str():
343 return {"type": "string", "default": ""}
344 case int():
345 return {"type": "int", "default": 0}
346 case Sequence():
347 tmp = {self._python2Avro(item)["type"] for item in value}
348 if len(tmp) == 0:
349 return {}
350 if len(tmp) > 1:
351 log.error(
352 "Sequence contains mixed types: %s, must be homogeneous for avro conversion "
353 "skipping record",
354 tmp,
355 )
356 return {"error_in_conversion": True}
357 return {"type": "array", "items": tmp.pop()}
358 case _:
359 log.error("Unsupported type %s, skipping record", type(value))
360 return {}
362 def _handleReferencePackage(self, meta: MutableMapping, bundle: MetricMeasurementBundle) -> None:
363 """Check to see if there is a reference package.
365 if there is a reference package, determine the datetime associated with
366 this reference package. Save the package, the version, and the date to
367 the common metric fields.
369 Parameters
370 ----------
371 meta : `MutableMapping`
372 A mapping which corresponds to fields which should be encoded in
373 all records.
374 bundle : `MetricMeasurementBundle`
375 The bundled metrics
376 """
377 ref_package, package_version, package_timestamp = "", "", 0.0
378 # This should only attempt to resolve a reference package if
379 # it is to be used to determine the timestamp.
380 if getattr(bundle, "timestamp_version", None) == "reference_package_timestamp":
381 if ref_package := getattr(bundle, "reference_package", ""):
382 packages = getEnvironmentPackages(True)
383 if package_info := packages.get(ref_package):
384 try:
385 package_version, package_timestamp = _tag2VersionTime(package_info)
386 except ValueError:
387 # Could not extract package timestamp leaving empty
388 pass
389 # explicit handle if None was set in the bundle for the package
390 meta["reference_package"] = ref_package or ""
391 meta["reference_package_version"] = package_version
392 meta["reference_package_timestamp"] = package_timestamp
394 def _handleTimes(self, meta: MutableMapping, bundle: MetricMeasurementBundle, run: str) -> None:
395 """Add times to the meta fields mapping.
397 Add all appropriate timestamp fields to the meta field mapping. These
398 will be added to all records.
400 This method will also look at the bundle to see if it defines a
401 preferred time. It so it sets that time as the main time stamp to be
402 used for this record.
404 Parameters
405 ----------
406 meta : `MutableMapping`
407 A mapping which corresponds to fields which should be encoded in
408 all records.
409 bundle : `MetricMeasurementBundle`
410 The bundled metrics
411 run : `str`
412 The `~lsst.daf.butler.Butler` collection where the
413 `MetricMeasurementBundle` is stored.
414 """
415 # Determine the timestamp associated with the run, if someone abused
416 # the run collection, use the current timestamp
417 if re.match(r"\d{8}T\d{6}Z", stamp := run.split("/")[-1]): 417 ↛ 418line 417 didn't jump to line 418 because the condition on line 417 was never true
418 run_timestamp = datetime.datetime.strptime(stamp, r"%Y%m%dT%H%M%S%z")
419 else:
420 run_timestamp = datetime.datetime.now()
421 meta["run_timestamp"] = run_timestamp.timestamp()
423 # If the bundle supports supplying timestamps, dispatch on the type
424 # specified.
425 if hasattr(bundle, "timestamp_version") and bundle.timestamp_version: 425 ↛ 458line 425 didn't jump to line 458 because the condition on line 425 was always true
426 match bundle.timestamp_version:
427 case "reference_package_timestamp": 427 ↛ 428line 427 didn't jump to line 428 because the pattern on line 427 never matched
428 if not meta["reference_package_timestamp"]:
429 log.error("Reference package timestamp is empty, using run_timestamp")
430 meta["timestamp"] = meta["run_timestamp"]
431 else:
432 meta["timestamp"] = meta["reference_package_timestamp"]
433 case "run_timestamp": 433 ↛ 434line 433 didn't jump to line 434 because the pattern on line 433 never matched
434 meta["timestamp"] = meta["run_timestamp"]
435 case "current_timestamp": 435 ↛ 436line 435 didn't jump to line 436 because the pattern on line 435 never matched
436 timeStamp = datetime.datetime.now()
437 meta["timestamp"] = timeStamp.timestamp()
438 case "dataset_timestamp": 438 ↛ 439line 438 didn't jump to line 439 because the pattern on line 438 never matched
439 log.error("dataset timestamps are not yet supported, run_timestamp will be used")
440 meta["timestamp"] = meta["run_timestamp"]
441 case str(value) if "explicit_timestamp" in value: 441 ↛ 450line 441 didn't jump to line 450 because the pattern on line 441 always matched
442 try:
443 _, splitTime = value.split(":")
444 except ValueError as excpt:
445 raise ValueError(
446 "Explicit timestamp must be given in the format 'explicit_timestamp:datetime', "
447 "where datetime is given in the form '%Y%m%dT%H%M%S%z"
448 ) from excpt
449 meta["timestamp"] = datetime.datetime.strptime(splitTime, r"%Y%m%dT%H%M%S%z").timestamp()
450 case _:
451 log.error(
452 "Timestamp version %s is not supported, run_timestamp will be used",
453 bundle.timestamp_version,
454 )
455 meta["timestamp"] = meta["run_timestamp"]
456 # Default to using the run_timestamp.
457 else:
458 meta["timestamp"] = meta["run_timestamp"]
460 def _handleIdentifier(
461 self,
462 meta: MutableMapping,
463 identifierFields: Mapping[str, Any] | None,
464 datasetIdentifier: str | None,
465 bundle: MetricMeasurementBundle,
466 ) -> None:
467 """Add an identifier to the meta record mapping.
469 If the bundle declares a dataset identifier to use add that to the
470 record, otherwise use 'Generic' as the identifier. If the
471 datasetIdentifier parameter is specified, that is used instead of
472 anything specified by the bundle.
474 This will also add any identifier fields supplied to the meta record
475 mapping.
477 Together these values (in addition to the timestamp and topic) should
478 uniquely identify an upload to the Sasquatch system.
480 Parameters
481 ----------
482 meta : `MutableMapping`
483 A mapping which corresponds to fields which should be encoded in
484 all records.
485 identifierFields: `Mapping` or `None`
486 The keys and values in this mapping will be both added as fields
487 in the record, and used in creating a unique tag for the uploaded
488 dataset type. I.e. the timestamp, and the tag will be unique, and
489 each record will belong to one combination of such.
490 datasetIdentifier : `str` or `None`
491 A string which will be used in creating unique identifier tags.
492 bundle : `MetricMeasurementBundle`
493 The bundle containing metric values to upload.
494 """
495 identifier: str
496 if datasetIdentifier is not None:
497 identifier = datasetIdentifier
498 elif hasattr(bundle, "dataset_identifier") and bundle.dataset_identifier is not None:
499 identifier = bundle.dataset_identifier
500 else:
501 identifier = "Generic"
503 meta["dataset_tag"] = identifier
505 if identifierFields is None:
506 identifierFields = {}
507 for key in IDENTIFIER_KEYS:
508 value = identifierFields.get(key, "")
509 meta[key] = f"{value}"
511 def _prepareBundle(
512 self,
513 bundle: MetricMeasurementBundle,
514 run: str,
515 datasetType: str,
516 timestamp: datetime.datetime | None = None,
517 id: UUID | None = None,
518 identifierFields: Mapping | None = None,
519 datasetIdentifier: str | None = None,
520 extraFields: Mapping | None = None,
521 ) -> tuple[Mapping[str, list[Any]], bool]:
522 """Encode all of the inputs into a format that can be sent to the
523 kafka proxy server.
525 Parameters
526 ----------
527 bundle : `MetricMeasurementBundle`
528 The bundle containing metric values to upload.
529 run : `str`
530 The run name to associate with these metric values. If this bundle
531 is also stored in the butler, this should be the butler run
532 collection the bundle is stored in the butler.
533 datasetType : `str`
534 The dataset type name associated with this
535 `MetricMeasurementBundle`
536 timestamp : `datetime.datetime`, optional
537 The timestamp to be associated with the measurements in the ingress
538 database. If this value is None, timestamp will be set by the run
539 time or current time.
540 id : `UUID`, optional
541 The UUID of the `MetricMeasurementBundle` within the butler. If
542 `None`, a new random UUID will be generated so that each record in
543 Sasquatch will have a unique value.
544 identifierFields: `Mapping`, optional
545 The keys and values in this mapping will be both added as fields
546 in the record, and used in creating a unique tag for the uploaded
547 dataset type. I.e. the timestamp, and the tag will be unique, and
548 each record will belong to one combination of such.
549 datasetIdentifier : `str`, optional
550 A string which will be used in creating unique identifier tags.
551 extraFields: `Mapping`, optional
552 Extra mapping keys and values that will be added as fields to the
553 dispatched record.
555 Returns
556 -------
557 result : `Mapping` of `str` to `list`
558 A mapping of metric name of list of metric measurement records.
559 status : `bool`
560 A status boolean indicating if some records had to be skipped due
561 to a problem parsing the bundle.
562 """
563 if id is None:
564 id = uuid4()
565 sid = str(id)
566 meta: dict[str, Any] = dict()
568 # Add other associated common fields
569 meta["id"] = sid
570 meta["run"] = run
571 meta["dataset_type"] = datasetType
573 # Check to see if the bundle declares a reference package
574 self._handleReferencePackage(meta, bundle)
576 # Handle the various timestamps that could be associated with a record
577 self._handleTimes(meta, bundle, run)
579 # Always use the supplied timestamp if one was passed to use.
580 if timestamp is not None:
581 meta["timestamp"] = timestamp.timestamp()
583 self._handleIdentifier(meta, identifierFields, datasetIdentifier, bundle)
585 # Add in any other fields that were supplied to the function call.
586 if extraFields is not None:
587 meta.update(extraFields)
589 metricRecords: dict[str, list[Any]] = dict()
591 # Record if any records needed skipped
592 resultsTrimmed = False
594 # Look at each of the metrics in the bundle (name, values)
595 for metric, measurements in bundle.items():
596 # Create a list which will contain the records for each measurement
597 # associated with metric.
598 metricRecordList = metricRecords.setdefault(f"{bundle.metricNamePrefix}{metric}", list())
600 record: dict[str, Any] = meta.copy()
602 # loop over each metric measurement within the metric
603 for measurement in measurements:
604 # need to extract any tags, package info, etc
605 note_key = f"{measurement.metric_name.metric}.metric_tags"
606 record["tags"] = dict(measurement.notes.items()).get(note_key, list())
608 # Missing values are replaced by 0 in sasquatch, see RFC-763.
609 name = ""
610 value = 0.0
611 match measurement.json:
612 case {"metric": name, "value": None}:
613 pass
614 case {"metric": name, "value": value}:
615 if math.isnan(value):
616 log.error(
617 "Measurement %s had a value that is a NaN, dispatch will be skipped",
618 measurement,
619 )
620 resultsTrimmed = True
621 continue
622 # JSON will not serialize np.float32, must cast.
623 if isinstance(value, np.float32):
624 value = float(value)
625 pass
626 case {"value": _}:
627 log.error("Measurement %s does not contain the key 'metric'", measurement)
628 resultsTrimmed = True
629 continue
630 case {"metric": _}:
631 log.error("Measurement %s does not contain the key 'value'", measurement)
632 resultsTrimmed = True
633 continue
634 record[name] = value
636 metricRecordList.append({"value": record})
637 return metricRecords, resultsTrimmed
639 def dispatch(
640 self,
641 bundle: MetricMeasurementBundle,
642 run: str,
643 datasetType: str,
644 timestamp: datetime.datetime | None = None,
645 id: UUID | None = None,
646 datasetIdentifier: str | None = None,
647 identifierFields: Mapping | None = None,
648 extraFields: Mapping | None = None,
649 ) -> None:
650 """Dispatch a `MetricMeasurementBundle` to Sasquatch.
652 Parameters
653 ----------
654 bundle : `MetricMeasurementBundle`
655 The bundle containing metric values to upload.
656 run : `str`
657 The run name to associate with these metric values. If this bundle
658 is also stored in the butler, this should be the butler run
659 collection the bundle is stored in the butler. This will be used
660 in generating uniqueness constraints in Sasquatch.
661 datasetType : `str`
662 The dataset type name associated with this
663 `MetricMeasurementBundle`.
664 timestamp : `datetime.datetime`, optional
665 The timestamp to be associated with the measurements in the ingress
666 database. If this value is None, timestamp will be set by the run
667 time or current time.
668 id : `UUID`, optional
669 The UUID of the `MetricMeasurementBundle` within the Butler. If
670 `None`, a new random UUID will be generated so that each record in
671 Sasquatch will have a unique value.
672 datasetIdentifier : `str`, optional
673 A string which will be used in creating unique identifier tags. If
674 `None`, a default value will be inserted.
675 identifierFields: `Mapping`, optional
676 The keys and values in this mapping will be both added as fields
677 in the record, and used in creating a unique tag for the uploaded
678 dataset type. I.e. the timestamp, and the tag will be unique, and
679 each record will belong to one combination of such. Examples of
680 entries would be things like visit or tract.
681 extraFields: `Mapping`, optional
682 Extra mapping keys and values that will be added as fields to the
683 dispatched record.
685 Raises
686 ------
687 SasquatchDispatchPartialFailure, SasquatchDispatchFailure
688 Raised if there were any errors in dispatching a bundle.
689 """
690 if id is None:
691 id = uuid4()
693 # Prepare the bundle by transforming it to a list of metric records
694 metricRecords, recordsTrimmed = self._prepareBundle(
695 bundle=bundle,
696 run=run,
697 datasetType=datasetType,
698 timestamp=timestamp,
699 id=id,
700 datasetIdentifier=datasetIdentifier,
701 identifierFields=identifierFields,
702 extraFields=extraFields,
703 )
705 headers = {
706 "content-type": "application/vnd.kafka.avro.v2+json",
707 "Authorization": f"Bearer {self.token}",
708 }
709 data: dict[str, Any] = dict()
710 partialUpload = False
711 uploadFailed = []
713 with http_client() as session:
714 for metric, record in metricRecords.items():
715 # create the kafka topic if it does not already exist
716 if not self._create_topic(metric):
717 log.error("Topic not created, skipping dispatch of %s", metric)
718 continue
719 recordValue = record[0]["value"]
720 # Generate schemas for each record
721 data["value_schema"], schemaTrimmed = self._generateAvroSchema(metric, recordValue)
722 data["records"] = record
724 if schemaTrimmed:
725 partialUpload = True
727 try:
728 r = session.post(
729 f"{self.url}/topics/{self.namespace}.{metric}",
730 json=data,
731 headers=headers,
732 timeout=self.timeout,
733 )
734 r.raise_for_status()
735 log.debug("Succesfully sent data for metric %s", metric)
736 uploadFailed.append(False)
737 except requests.HTTPError as e:
738 log.error(
739 "There was a problem submitting the metric %s: %s, %s",
740 metric,
741 e.response.status_code,
742 e.response.reason,
743 )
744 uploadFailed.append(True)
745 partialUpload = True
746 except requests.RequestException as e:
747 # Don't log full stack trace because there may be lots
748 # of these.
749 log.error("There was a problem submitting the metric %s: %s", metric, e)
750 uploadFailed.append(True)
751 partialUpload = True
753 # There may be no metrics to try to upload, and thus the uploadFailed
754 # list may be empty, check before issuing failure
755 if len(uploadFailed) > 0 and all(uploadFailed):
756 raise SasquatchDispatchFailure("All records were unable to be uploaded.")
758 if partialUpload or recordsTrimmed:
759 raise SasquatchDispatchPartialFailure("One or more records may not have been uploaded entirely")
761 def dispatchRef(
762 self,
763 bundle: MetricMeasurementBundle,
764 ref: DatasetRef,
765 timestamp: datetime.datetime | None = None,
766 extraFields: Mapping | None = None,
767 datasetIdentifier: str | None = None,
768 ) -> None:
769 """Dispatch a `MetricMeasurementBundle` to Sasquatch with a known
770 `DatasetRef`.
772 Parameters
773 ----------
774 bundle : `MetricMeasurementBundle`
775 The bundle containing metric values to upload.
776 ref : `DatasetRef`
777 The `Butler` dataset ref corresponding to the input
778 `MetricMeasurementBundle`.
779 timestamp : `datetime.datetime`, optional
780 The timestamp to be associated with the measurements in the ingress
781 database. If this value is None, timestamp will be set by the run
782 time or current time.
783 extraFields: `Mapping`, optional
784 Extra mapping keys and values that will be added as fields to the
785 dispatched record if not None.
786 datasetIdentifier : `str`, optional
787 A string which will be used in creating unique identifier tags. If
788 None, a default value will be inserted.
790 Raises
791 ------
792 SasquatchDispatchPartialFailure, SasquatchDispatchFailure
793 Raised if there were any errors in dispatching a bundle.
794 """
795 # Parse the relevant info out of the dataset ref.
796 serializedRef = ref.to_simple()
797 id = serializedRef.id
798 if serializedRef.run is None:
799 run = "<unknown>"
800 else:
801 run = serializedRef.run
802 dstype = serializedRef.datasetType
803 datasetType = dstype.name if dstype is not None else ""
804 dataRefMapping = serializedRef.dataId.dataId if serializedRef.dataId else None
806 self.dispatch(
807 bundle,
808 run=run,
809 timestamp=timestamp,
810 datasetType=datasetType,
811 id=id,
812 identifierFields=dataRefMapping,
813 extraFields=extraFields,
814 datasetIdentifier=datasetIdentifier,
815 )