This repository has been archived by the owner on Oct 15, 2022. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 100
[Issue 81] Python 3 version of Statsite Sink and Vagrant update #82
Open
hkp
wants to merge
2
commits into
obfuscurity:master
Choose a base branch
from
hkp:master
base: master
Could not load branches
Branch not found: {{ refName }}
Loading
Could not load tags
Nothing to show
Loading
Are you sure you want to change the base?
Some commits from the old base branch may be removed from the timeline,
and old review comments may become outdated.
Open
Changes from all commits
Commits
Show all changes
2 commits
Select commit
Hold shift + click to select a range
File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,186 @@ | ||
| """ | ||
| Supports flushing metrics to graphite | ||
| """ | ||
| import re | ||
| import sys | ||
| import socket | ||
| import logging | ||
| import pickle | ||
| import struct | ||
| from builtins import range | ||
|
|
||
| # Initialize the logger | ||
| logging.basicConfig() | ||
|
|
||
| SPACES = re.compile(r"\s+") | ||
| SLASHES = re.compile(r"\/+") | ||
| NON_ALNUM = re.compile(r"[^a-zA-Z_\-0-9\.]") | ||
|
|
||
|
|
||
| class GraphiteStore(object): | ||
| def __init__(self, host="localhost", port=2003, prefix="statsite.", attempts=3, | ||
| protocol='lines', normalize=None, socket_timeout=2): | ||
| """ | ||
| Implements an interface that allows metrics to be persisted to Graphite. | ||
| Raises a :class:`ValueError` on bad arguments. | ||
|
|
||
| :Parameters: | ||
| - `host` : The hostname of the graphite server. | ||
| - `port` : The port of the graphite server | ||
| - `prefix` (optional) : A prefix to add to the keys. Defaults to 'statsite.' | ||
| - `attempts` (optional) : The number of re-connect retries before failing. | ||
| - `normalize` (optional) : If set, attempt to sanitize/normalize keys to be more | ||
| generally compliant with graphite/carbon expectations. | ||
| """ | ||
| # Convert the port to an int since its coming from a configuration file | ||
| port = int(port) | ||
| attempts = int(attempts) | ||
|
|
||
| if port <= 0: | ||
| raise ValueError("Port must be positive!") | ||
| if attempts < 1: | ||
| raise ValueError("Must have at least 1 attempt!") | ||
| if protocol not in ["pickle", "lines"]: | ||
| raise ValueError("Supported protocols are pickle, lines") | ||
|
|
||
| if normalize is not None and normalize not in ("False", "false", "No", "no"): | ||
| self.normalize_func = self.normalize_key | ||
| else: | ||
| self.normalize_func = lambda k: "%s%s" % (self.prefix, k) | ||
|
|
||
| self.logger = logging.getLogger("statsite.graphitestore") | ||
| self.host = host | ||
| self.port = port | ||
| self.prefix = prefix | ||
| self.attempts = attempts | ||
| self.socket_timeout = None if socket_timeout == "infinity" else socket_timeout | ||
| self.sock = self._create_socket() | ||
| self.flush = self.flush_pickle if protocol == "pickle" else self.flush_lines | ||
| self.metrics = [] | ||
|
|
||
| def normalize_key(self, key): | ||
| """ | ||
| Take a single key string and return the same string with spaces, slashes and | ||
| non-alphanumeric characters subbed out and prefixed by self.prefix. | ||
| """ | ||
| key = SPACES.sub("_", key) | ||
| key = SLASHES.sub("-", key) | ||
| key = NON_ALNUM.sub("", key) | ||
| key = "%s%s" % (self.prefix, key) | ||
| return key | ||
|
|
||
| def append(self, metric): | ||
| """ | ||
| Add one metric to queue for sending. Addtionally modify key to be compatible with txstatsd | ||
| format. | ||
|
|
||
| :Parameters: | ||
| - `metric` : A single statsd metric string in the format "key|value|timestamp". | ||
| """ | ||
| if metric and metric.count("|") == 2: | ||
| k, v, ts = metric.split("|") | ||
| k = self.normalize_func(k) | ||
| self.metrics.append(((k), v, ts)) | ||
|
|
||
| def send_metrics(self): | ||
| self.logger.info("Outputting %d metrics", len(self.metrics)) | ||
| self.flush() | ||
| self.metrics = [] | ||
|
|
||
| def flush_lines(self): | ||
| """ | ||
| Flushes the metrics provided to Graphite. | ||
| """ | ||
| if not self.metrics: | ||
| return | ||
|
|
||
| lines = ["%s %s %s" % metric for metric in self.metrics] | ||
| data = "\n".join(lines) + "\n" | ||
|
|
||
| # Serialize writes to the socket | ||
| try: | ||
| self._write_metric(data) | ||
| except ValueError: | ||
| self.logger.exception("Failed to write out the metrics!") | ||
|
|
||
| def flush_pickle(self): | ||
| """ | ||
| Flushes the metrics provided to Graphite. | ||
| """ | ||
| if not self.metrics: | ||
| return | ||
|
|
||
| # transform a list of strings into the list of tuples that | ||
| # pickle graphite interface supports, in the form of | ||
| # (key, (timestamp, value)) | ||
| # http://graphite.readthedocs.io/en/latest/feeding-carbon.html#the-pickle-protocol | ||
| metrics_fmt = [] | ||
| for (k, v, ts) in self.metrics: | ||
| metrics_fmt.append((k, (ts, v))) | ||
|
|
||
| # do pickle the list of tuples | ||
| # add the header the pickle protocol wants | ||
| payload = pickle.dumps(metrics_fmt, protocol=2) | ||
| header = struct.pack("!L", len(payload)) | ||
| message = header + payload | ||
|
|
||
| try: | ||
| self._write_metric(message) | ||
| except StandardError: | ||
| self.logger.exception("Failed to write out the metrics!") | ||
|
|
||
| def close(self): | ||
| """ | ||
| Closes the connection. The socket will be recreated on the next | ||
| flush. | ||
| """ | ||
| try: | ||
| if self.sock: | ||
| self.sock.close() | ||
| except StandardError: | ||
| self.logger.warning("Failed to close connection!") | ||
|
|
||
| def _create_socket(self): | ||
| """Creates a socket and connects to the graphite server""" | ||
| sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) | ||
| sock.settimeout(self.socket_timeout) | ||
| try: | ||
| sock.connect((self.host, self.port)) | ||
| except ValueError: | ||
| self.logger.error("Failed to connect!") | ||
| sock = None | ||
| return sock | ||
|
|
||
| def _write_metric(self, metric): | ||
| """Tries to write a string to the socket, reconnecting on any errors""" | ||
| for _ in range(self.attempts): | ||
| if self.sock: | ||
| try: | ||
| self.sock.sendall(metric.encode()) | ||
| return | ||
| except socket.error: | ||
| self.logger.exception("Error while flushing to graphite. Reattempting...") | ||
|
|
||
| self.sock = self._create_socket() | ||
|
|
||
| self.logger.critical("Failed to flush to Graphite! Gave up after %d attempts.", | ||
| self.attempts) | ||
|
|
||
|
|
||
| def main(): | ||
| # Intialize from our arguments | ||
| graphite = GraphiteStore(*sys.argv[1:]) | ||
|
|
||
| METRICS_PER_FLUSH = 5000 | ||
|
|
||
| # Get all the inputs | ||
| for line in sys.stdin: | ||
| if len(graphite.metrics) >= METRICS_PER_FLUSH: | ||
| graphite.send_metrics() | ||
| graphite.append(line.strip()) | ||
|
|
||
| graphite.send_metrics() | ||
|
|
||
|
|
||
| if __name__ == "__main__": | ||
| main() |
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Are you sure that section is needed?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This was for a code interview, so this is not needed for most people. I was setting up a multi-node environment and needed that variable so the nodes would know where to send the stats data.