|
14 | 14 | limitations under the License.""" |
15 | 15 |
|
16 | 16 | import sys |
17 | | -import imp |
18 | 17 | from os.path import dirname, join, abspath, exists |
19 | 18 | from optparse import OptionParser |
20 | 19 |
|
|
35 | 34 | except ImportError: |
36 | 35 | pass |
37 | 36 |
|
38 | | -from twisted.internet import stdio, reactor, defer |
39 | | -from twisted.protocols.basic import LineReceiver |
40 | | -from carbon.routers import ConsistentHashingRouter, RelayRulesRouter |
41 | | -from carbon.client import CarbonClientManager |
42 | | -from carbon import log, events |
| 37 | +from twisted.internet import stdio, reactor, defer # noqa |
| 38 | +from twisted.protocols.basic import LineReceiver # noqa |
| 39 | +from carbon.routers import ConsistentHashingRouter, RelayRulesRouter # noqa |
| 40 | +from carbon.client import CarbonClientManager # noqa |
| 41 | +from carbon import log, events # noqa |
43 | 42 |
|
44 | 43 |
|
45 | 44 | option_parser = OptionParser(usage="%prog [options] <host:port:instance> <host:port:instance> ...") |
46 | 45 | option_parser.add_option('--debug', action='store_true', help="Log debug info to stdout") |
47 | 46 | option_parser.add_option('--keyfunc', help="Use a custom key function (path/to/module.py:myFunc)") |
48 | 47 | option_parser.add_option('--replication', type='int', default=1, help='Replication factor') |
49 | | -option_parser.add_option('--routing', default='consistent-hashing', |
| 48 | +option_parser.add_option( |
| 49 | + '--routing', default='consistent-hashing', |
50 | 50 | help='Routing method: "consistent-hashing" (default) or "relay"') |
51 | | -option_parser.add_option('--diverse-replicas', action='store_true', help="Spread replicas across diff. servers") |
52 | | -option_parser.add_option('--relayrules', default=default_relayrules, |
53 | | - help='relay-rules.conf file to use for relay routing') |
| 51 | +option_parser.add_option( |
| 52 | + '--diverse-replicas', action='store_true', help="Spread replicas across diff. servers") |
| 53 | +option_parser.add_option( |
| 54 | + '--relayrules', default=default_relayrules, help='relay-rules.conf file to use for relay routing') |
54 | 55 |
|
55 | 56 | options, args = option_parser.parse_args() |
56 | 57 |
|
|
74 | 75 | instance = parts[2] |
75 | 76 | else: |
76 | 77 | instance = None |
77 | | - destinations.append( (host, port, instance) ) |
| 78 | + destinations.append((host, port, instance)) |
78 | 79 |
|
79 | 80 | if options.debug: |
80 | 81 | log.logToStdout() |
@@ -104,26 +105,31 @@ class StdinMetricsReader(LineReceiver): |
104 | 105 | delimiter = '\n' |
105 | 106 |
|
106 | 107 | def lineReceived(self, line): |
107 | | - #log.msg("[DEBUG] lineReceived(): %s" % line) |
| 108 | + # log.msg("[DEBUG] lineReceived(): %s" % line) |
108 | 109 | try: |
109 | 110 | (metric, value, timestamp) = line.split() |
110 | 111 | datapoint = (float(timestamp), float(value)) |
111 | | - assert datapoint[1] == datapoint[1] # filter out NaNs |
| 112 | + assert datapoint[1] == datapoint[1] # filter out NaNs |
112 | 113 | client_manager.sendDatapoint(metric, datapoint) |
113 | 114 | except ValueError: |
114 | 115 | log.err(None, 'Dropping invalid line: %s' % line) |
115 | 116 |
|
116 | 117 | def connectionLost(self, reason): |
117 | 118 | log.msg('stdin disconnected') |
| 119 | + |
118 | 120 | def startShutdown(results): |
119 | 121 | log.msg("startShutdown(%s)" % str(results)) |
120 | 122 | allStopped = client_manager.stopAllClients() |
121 | 123 | allStopped.addCallback(shutdown) |
| 124 | + |
122 | 125 | firstConnectsAttempted.addCallback(startShutdown) |
123 | 126 |
|
124 | | -stdio.StandardIO( StdinMetricsReader() ) |
| 127 | + |
| 128 | +stdio.StandardIO(StdinMetricsReader()) |
125 | 129 |
|
126 | 130 | exitCode = 0 |
| 131 | + |
| 132 | + |
127 | 133 | def shutdown(results): |
128 | 134 | global exitCode |
129 | 135 | for success, result in results: |
|
0 commit comments