netdata-timescale-relay: comparison netdata

equal deleted inserted replaced

-:a135fdaed52b
+:1f09cfb560e0
 # vim: set et nosta sw=4 ts=4 :
 #
-# Copyright (c) 2018, Mahlon E. Smith <mahlon@martini.nu>
+# Copyright (c) 2018-2020, Mahlon E. Smith <mahlon@martini.nu>
 # All rights reserved.
 # Redistribution and use in source and binary forms, with or without
 # modification, are permitted provided that the following conditions are met:
 #
 #     * Redistributions of source code must retain the above copyright
 db_postgres,
 json,
 math,
 nativesockets,
 net,
-os,
 parseopt,
 posix,
 strutils,
+strformat,
 tables,
 terminal,
 times
 const
-VERSION = "v0.2.0"
+VERSION = "v0.3.0"
 USAGE = """
-./netdata_tsrelay [-d][-h][-q][-t][-T][-v] --dbopts="[PostgreSQL connection string]" --listen-port=14866 --listen-addr=0.0.0.0
+./netdata_tsrelay [-adDhopqtTv]
--q: Quiet mode.  No output at all.  Ignored if -d is supplied.
+-a --listen-addr:
--d: Debug: Show incoming and parsed data.
+The outbound IP address to listen for netdata streams.
--v: Display version number.
--T: Change the destination table name from the default 'netdata'.
+-d --debug:
--t: Alter the maximum time (in ms) an open socket waits for data.  Default: 500ms.
+Debug: Show incoming and parsed data.
--h: Help.  You're lookin' at it.
+-D --dropconn:
-The default connection string is:
+Drop the persistent socket to netdata between samples to conserve
-"host=localhost dbname=netdata application_name=netdata-tsrelay"
+local resources.  This may be helpful with a large number of clients.
+Defaults to false.
+-h --help:
+Help.  You're lookin' at it.
+-o --dbopts:
+The PostgreSQL connection string parameters.
+The default connection string is:
+"host=localhost dbname=netdata application_name=netdata-tsrelay"
+-p --listen-port:
+Change the listening port from the default (14866).
+-P --persistent:
+Don't disconnect from the database between samples.  This may be
+more efficient with a small number of clients, when not using a
+pooler, or with a very high sample size/rate.  Defaults to false.
+-q --quiet:
+Quiet mode.  No output at all.  Ignored if -d is supplied.
+-T --dbtable:
+Change the destination table name from the default (netdata).
+-t --timeout:
+Alter the maximum time (in ms) an open socket waits for data
+before processing the sample.  Default: 500ms.
+-v --verbose:
+Display version number.
 """
 INSERT_SQL = """
 INSERT INTO $1
 ( time, host, metrics )
 VALUES
 type
 Config = object of RootObj
 dbopts:      string  # The postgresql connection parameters.  (See https://www.postgresql.org/docs/current/static/libpq-connect.html)
 dbtable:     string  # The name of the table to write to.
+dropconn:    bool    # Close the TCP connection between samples.
+persistent:  bool    # Don't close the database handle between samples.
 listen_port: int     # The port to listen for incoming connections.
 listen_addr: string  # The IP address listen for incoming connections.  Defaults to inaddr_any.
 verbose:     bool    # Be informative
 debug:       bool    # Spew out raw data
 insertsql:   string  # The SQL insert string after interpolating the table name.
 timeout:     int     # How long to block, waiting on connection data.
+type
+NetdataClient = ref object
+sock: Socket    # The raw socket fd
+address: string # The remote IP address
+db: DbConn      # An optionally persistent database handle
 # Global configuration
 var conf: Config
 proc hl( msg: string, fg: ForegroundColor, bright=false ): string =
 var color: BiggestInt = ord( fg )
 if bright: inc( color, 60 )
 result = "\e[" & $color & 'm' & msg & "\e[0m"
-proc fetch_data( client: Socket ): string =
+proc fetch_data( client: NetdataClient ): string =
-## Netdata JSON backend doesn't send a length, so we read line by
+## Netdata JSON backend doesn't send a length nor a separator
-## line and wait for stream timeout to determine a "sample".
+## between samples, so we read line by line and wait for stream
+## timeout to determine what constitutes a sample.
 var buf = ""
-try:
+while true:
-while true:
+try:
-client.readline( buf, timeout=conf.timeout )
+client.sock.readline( buf, timeout=conf.timeout )
-if buf != "": result = result & buf & "\n"
+if buf == "":
-except TimeoutError:
+if conf.debug: echo "Client {client.address} closed socket.".fmt.hl( fgRed, bright=true )
-return
+quit( 1 )
+result = result & buf & "\n"
+except OSError:
+quit( 1 )
+except TimeoutError:
+if result == "": continue
+return
 proc parse_data( data: string ): seq[ JsonNode ] =
 ## Given a raw +data+ string, parse JSON and return a sequence
 ## of JSON samples. Netdata can buffer multiple samples in one batch.
 for timestamp, sample in pivoted_data:
 result.add( sample )
-proc write_to_database( samples: seq[ JsonNode ] ): void =
+proc write_to_database( client: NetdataClient, samples: seq[ JsonNode ] ): void =
 ## Given a sequence of json samples, write them to database.
 if samples.len == 0: return
-let db = open( "", "", "", conf.dbopts )
+if client.db.isNil:
+client.db = open( "", "", "", conf.dbopts )
 try:
-db.exec sql( "BEGIN" )
+client.db.exec sql( "BEGIN" )
 for sample in samples:
 var
 timestamp = sample[ "timestamp" ].get_int
 host = sample[ "hostname" ].get_str.to_lowerascii
 sample.delete( "timestamp" )
 sample.delete( "hostname" )
-db.exec sql( conf.insertsql ), timestamp, host, sample
+client.db.exec sql( conf.insertsql ), timestamp, host, sample
-db.exec sql( "COMMIT" )
+client.db.exec sql( "COMMIT" )
 except:
 let
 e = getCurrentException()
 msg = getCurrentExceptionMsg()
 echo "Got exception ", repr(e), " while writing to DB: ", msg
 discard
-db.close
+if not conf.persistent:
+client.db.close
+client.db = nil
-proc process( client: Socket, address: string ): void =
+proc process( client: NetdataClient ): void =
 ## Do the work for a connected client within child process.
 let t0 = cpu_time()
 var raw_data = client.fetch_data
 # Done with the socket, netdata will automatically
 # reconnect.  Save local resources/file descriptors
 # by closing after the send is considered complete.
 #
-try:
+if conf.dropconn:
-client.close
+try:
-except OSError:
+client.sock.close
-return
+except OSError:
+return
 # Pivot the parsed data to a single JSON blob per sample time.
 var samples = parse_data( raw_data )
-write_to_database( samples )
+client.write_to_database( samples )
 if conf.verbose:
+let cputime = cpu_time() - t0
 echo(
 hl( $(epochTime().to_int), fgMagenta, bright=true ),
 " ",
 hl( $(samples.len), fgWhite, bright=true ),
 " sample(s) parsed from ",
-address.hl( fgYellow, bright=true ),
+client.address.hl( fgYellow, bright=true ),
-" in ", hl($( round(cpu_time() - t0, 3) ), fgWhite, bright=true), " seconds."
+" in ", hl( "{cputime:<2.3f}".fmt, fgWhite, bright=true), " seconds."
-# " ", hl($(round((get_occupied_mem()/1024/1024),1)), fgWhite, bright=true), "MB memory used."
 )
 proc serverloop( conf: Config ): void =
 ## Open a database connection, bind to the listening socket,
 echo ""
 # Wait for incoming connections, fork for each client.
 #
 while true:
-var
+let client = NetdataClient.new
-client  = new Socket
+client.sock = Socket.new
-address = ""
 # Block, waiting for new connections.
-server.acceptAddr( client, address )
+server.acceptAddr( client.sock, client.address )
 if fork() == 0:
 server.close
-client.process( address )
+if conf.dropconn:
-quit( 0 )
+# "one shot" mode.
+client.process
-client.close
+quit( 0 )
+else:
+# Keep the connection to netdata open.
+while true: client.process
+client.sock.close
 when defined( testing ): dumpNumberOfInstances()
 proc parse_cmdline: Config =
 ## Populate the config object with the user's preferences.
 # Config object defaults.
 #
 result = Config(
 dbopts: "host=localhost dbname=netdata application_name=netdata-tsrelay",
 dbtable: "netdata",
+dropconn: false,
 listen_port: 14866,
 listen_addr: "0.0.0.0",
 verbose: true,
 debug: false,
 timeout: 500,
+persistent: false,
 insertsql: INSERT_SQL % [ "netdata" ]
 )
 # always set debug mode if development build.
 result.debug = defined( testing )
 of cmdLongOption, cmdShortOption:
 case key
 of "debug", "d":
 result.debug = true
+of "dropconn", "D":
+if result.persistent:
+echo "Dropping TCP sockets are incompatible with persistent database connections."
+quit( 1 )
+result.dropconn = true
 of "help", "h":
 echo USAGE
 quit( 0 )
 of "quiet", "q":
 of "timeout", "t": result.timeout = val.parse_int
 of "dbtable", "T":
 result.insertsql = INSERT_SQL % [ val ]
-of "dbopts": result.dbopts = val
+of "dbopts", "o": result.dbopts = val
 of "listen-addr", "a": result.listen_addr = val
 of "listen-port", "p": result.listen_port = val.parse_int
+of "persistent", "P":
+if result.dropconn:
+echo "Persistent database connections are incompatible with dropping TCP sockets."
+quit( 1 )
+result.persistent = true
 else: discard
 of cmdEnd: assert( false ) # shouldn't reach here ever

changeset 19	1f09cfb560e0
parent 17	96b8799a565a
child 21	a2fe9ec4cdf2