<s> import time <EOL> import json <EOL> import socket <EOL> import itertools <EOL> import urlparse <EOL> import requests <EOL> import logging <EOL> POINT_LIMIT = <NUM_LIT> <EOL> class Client ( object ) : <EOL> def __init__ ( self , host_query , host_write , port_query , port_write , db ) : <EOL> self . query_uri = urlparse . urlunparse ( urlparse . ParseResult ( <EOL> '<STR_LIT:http>' , '<STR_LIT>' % ( host_query , port_query ) , '<STR_LIT>' , <EOL> None , None , None ) ) <EOL> self . prefix = db <EOL> self . write_addr = ( host_write , port_write ) <EOL> self . socket = None <EOL> self . stream = None <EOL> self . id_counter = None <EOL> self . buf_size = None <EOL> self . reconnect ( ) <EOL> def reconnect ( self ) : <EOL> self . close ( ) <EOL> self . socket = socket . create_connection ( self . write_addr ) <EOL> self . stream = self . socket . makefile ( ) <EOL> self . id_counter = itertools . count ( ) <EOL> self . buf_size = <NUM_LIT:1> << <NUM_LIT:16> <EOL> def close ( self ) : <EOL> if self . socket is None : <EOL> return <EOL> self . socket . close ( ) <EOL> self . stream . close ( ) <EOL> def __del__ ( self ) : <EOL> self . close ( ) <EOL> def write_points ( self , name , fields ) : <EOL> now = int ( time . time ( ) ) <EOL> try : <EOL> self . _write ( [ { <EOL> '<STR_LIT>' : metric , <EOL> '<STR_LIT>' : self . prefix , <EOL> '<STR_LIT>' : now , <EOL> '<STR_LIT>' : <NUM_LIT:30> , <EOL> '<STR_LIT:value>' : val , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' + name , <EOL> } for metric , val in fields . iteritems ( ) ] ) <EOL> except IOError as e : <EOL> logging . error ( '<STR_LIT>' , name , e . message ) <EOL> self . reconnect ( ) <EOL> raise <EOL> def _write ( self , lines ) : <EOL> s = <NUM_LIT:0> <EOL> resp = [ ] <EOL> while True : <EOL> buf = lines [ s : s + self . buf_size ] <EOL> s = s + self . buf_size <EOL> if len ( buf ) == <NUM_LIT:0> : <EOL> break <EOL> r = self . _rpc ( '<STR_LIT>' , buf ) <EOL> resp . append ( r ) <EOL> return resp <EOL> def _rpc ( self , name , * params ) : <EOL> request = { <EOL> '<STR_LIT:id>' : next ( self . id_counter ) , <EOL> '<STR_LIT>' : list ( params ) , <EOL> '<STR_LIT>' : name , <EOL> } <EOL> payload = json . dumps ( request ) <EOL> self . socket . sendall ( payload ) <EOL> response = self . stream . readline ( ) <EOL> if not response : <EOL> raise IOError ( '<STR_LIT>' ) <EOL> response = json . loads ( response . decode ( '<STR_LIT:utf8>' ) ) <EOL> if response . get ( '<STR_LIT:error>' ) is not None : <EOL> raise IOError ( response . get ( '<STR_LIT:error>' ) ) <EOL> return response . get ( '<STR_LIT:result>' ) <EOL> def query ( self , node , field , aggf , span , end , interval ) : <EOL> r = requests . post ( self . query_uri , data = json . dumps ( { <EOL> '<STR_LIT:start>' : end - span , <EOL> '<STR_LIT:end>' : end , <EOL> '<STR_LIT>' : aggf , <EOL> '<STR_LIT>' : [ { <EOL> '<STR_LIT>' : self . prefix , <EOL> '<STR_LIT>' : field + '<STR_LIT>' + node , <EOL> } ] , <EOL> } ) ) . json ( ) [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> if r is None : <EOL> return [ ] <EOL> if len ( r ) > POINT_LIMIT : <EOL> r = r [ : : len ( r ) / POINT_LIMIT + <NUM_LIT:1> ] <EOL> return [ [ x [ '<STR_LIT>' ] , x [ '<STR_LIT:value>' ] ] <EOL> for x in r if x [ '<STR_LIT:value>' ] is not None ] <EOL> def query_request ( self , node , end ) : <EOL> r = requests . post ( self . query_uri , data = json . dumps ( { <EOL> '<STR_LIT:start>' : end - <NUM_LIT> , <EOL> '<STR_LIT:end>' : end , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ { <EOL> '<STR_LIT>' : self . prefix , <EOL> '<STR_LIT>' : field + '<STR_LIT>' + node , <EOL> } ] , <EOL> } ) ) . json ( ) [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> if r : <EOL> for i in range ( <NUM_LIT:1> , len ( r ) ) : <EOL> m = r [ - i ] <EOL> if m [ '<STR_LIT:value>' ] : <EOL> return m [ '<STR_LIT:value>' ] </s>
