netdata-timescale-relay: comparison netdata

equal deleted inserted replaced

-:c0bcf3bea772
+:1ef3f2d6d10e
-# vim: set et nosta sw=4 ts=4 ft=nim :
+# vim: set et nosta sw=4 ts=4 :
+# im: set et nosta sw=4 ts=4 ft=nim :
 #
 # Copyright (c) 2018, 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:
 math,
 nativesockets,
 net,
 os,
 parseopt2,
+posix,
 strutils,
 tables,
 terminal,
-times,
+times
-threadpool
 const
 VERSION = "v0.1.0"
 USAGE = """
 ./netdata_tsrelay [-q][-v][-h] --dbopts="[PostgreSQL connection string]" --listen-port=14866 --listen-addr=0.0.0.0
 -q: Quiet mode.  No output at all.  Ignored if -d is supplied.
--c: Suppress ANSI color output.
 -d: Debug: Show incoming and parsed data.
 -v: Display version number.
 -h: Help.  You're lookin' at it.
 The default connection string is:
 dbopts:      string  # The postgresql connection parameters.  (See https://www.postgresql.org/docs/current/static/libpq-connect.html)
 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
-use_color:   bool    # Pretty things up a little, probably want to disable this if debugging
+# Global configuration
+var conf: Config
-# The global config object
-#
-# FIXME:  Rather than pass this all over the
-# place, consider channels and createThread instead of spawn.
-#
-var conf = Config(
-dbopts: "host=localhost port=5432 dbname=netdata user=netdata application_name=netdata-tsrelay",
-listen_port: 14866,
-listen_addr: "0.0.0.0",
-verbose: true,
-debug: false,
-use_color: true
-)
 proc hl( msg: string, fg: ForegroundColor, bright=false ): string =
 ## Quick wrapper for color formatting a string, since the 'terminal'
 ## module only deals with stdout directly.
-if not conf.use_color: return msg
+if not isatty(stdout): return msg
 var color: BiggestInt = ord( fg )
 if bright: inc( color, 60 )
 result = "\e[" & $color & 'm' & msg & "\e[0m"
 proc fetch_data( client: Socket ): string =
 ## Netdata JSON backend doesn't send a length, so we read line by
 ## line and wait for stream timeout to determine a "sample".
+var buf: string = nil
 try:
-result = client.recv_line( timeout=500 ) & "\n"
+result = client.recv_line( timeout=500 )
-while result != "":
+if result != "" and not result.is_nil: result = result & "\n"
-result = result & client.recv_line( timeout=500 ) & "\n"
+while buf != "":
+buf = client.recv_line( timeout=500 )
+if buf != "" and not buf.is_nil: result = result & buf & "\n"
 except TimeoutError:
 discard
-proc parse_data( data: string, conf: Config ): Table[ BiggestInt, JsonNode ] =
+proc parse_data( data: string ): seq[ JsonNode ] =
-## Given a raw +data+ string, parse JSON and return a table of
+## Given a raw +data+ string, parse JSON and return a sequence
-## JSON samples ready for writing, keyed by timestamp. Netdata can
+## of JSON samples. Netdata can buffer multiple samples in one batch.
-## buffer multiple samples in one batch.
+result = @[]
-if data == "": return
+if data == "" or data.is_nil: return
 # Hash of sample timeperiods to pivoted json data
-result = init_table[ BiggestInt, JsonNode ]()
+var pivoted_data = init_table[ BiggestInt, JsonNode ]()
 for sample in split_lines( data ):
-if conf.debug: echo sample.hl( fgBlack, bright=true )
+if sample == "" or sample.is_nil: continue
-if sample.len == 0: continue
+#if conf.debug: echo sample.hl( fgBlack, bright=true )
 var parsed: JsonNode
 try:
 parsed = sample.parse_json
 except JsonParsingError:
-discard
 if conf.debug: echo hl( "Unable to parse sample line: " & sample.hl(fgRed, bright=true), fgRed )
+continue
+if parsed.kind != JObject: return
 # Create or use existing Json object for modded data.
 #
 var pivot: JsonNode
-let key = parsed["timestamp"].get_num
+try:
+let key = parsed["timestamp"].get_num
-if result.has_key( key ):
-pivot = result[ key ]
+if pivoted_data.has_key( key ):
-else:
+pivot = pivoted_data[ key ]
-pivot = newJObject()
+else:
-result[ key ] = pivot
+pivot = newJObject()
+pivoted_data[ key ] = pivot
-var name = parsed[ "chart_id" ].get_str & "." & parsed[ "id" ].get_str
-pivot[ "hostname" ] = parsed[ "hostname" ]
+var name = parsed[ "chart_id" ].get_str & "." & parsed[ "id" ].get_str
-pivot[ name ] = parsed[ "value" ]
+pivot[ "hostname" ] = parsed[ "hostname" ]
+pivot[ "timestamp" ] = parsed[ "timestamp" ]
-return result
+pivot[ name ] = parsed[ "value" ]
+except:
+continue
-proc process( client: Socket, db: DBConn, conf: Config ): int =
+for timestamp, sample in pivoted_data:
+result.add( sample )
+proc write_to_database( samples: seq[ JsonNode ] ): void =
+## Given a sequence of json samples, write them to database.
+if samples.len == 0: return
+let db = open( "", "", "", conf.dbopts )
+try:
+db.exec sql( "BEGIN" )
+for sample in samples:
+var
+timestamp = sample[ "timestamp" ].get_num
+host = sample[ "hostname" ].get_str
+sample.delete( "timestamp" )
+sample.delete( "hostname" )
+db.exec sql( INSERT_SQL ), timestamp, host, sample
+db.exec sql( "COMMIT" )
+except:
+let
+e = getCurrentException()
+msg = getCurrentExceptionMsg()
+echo "Got exception ", repr(e), " while writing to DB: ", msg
+discard
+db.close
+proc process( client: Socket, address: string ): void =
 ## Do the work for a connected client within a thread.
-## Returns the number of samples parsed.
+## Returns the formatted json data keyed on sample time.
+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:
 client.close
 except OSError:
 return
-# Pivot data and save to SQL.
+# Pivot the parsed data to a single JSON blob per sample time.
-#
+var samples = parse_data( raw_data )
-var samples = parse_data( raw_data, conf )
+write_to_database( samples )
-if samples.len != 0:
-db.exec sql( "BEGIN" )
-for timestamp, sample in samples:
-var host = sample[ "hostname" ].get_str
-sample.delete( "hostname" )
-db.exec sql( INSERT_SQL ), timestamp, host, sample
-db.exec sql( "COMMIT" )
-return samples.len
-proc runthread( client: Socket, address: string, db: DBConn, conf: Config ): void {.thread.} =
-## A thread that performs that dispatches processing and returns
-## results.
-let t0 = cpu_time()
-var samples = client.process( db, conf )
 if conf.verbose:
 echo(
-hl( $samples, fgWhite, bright=true ),
+hl( $(epochTime().to_int), fgMagenta, bright=true ),
+" ",
+hl( $(samples.len), fgWhite, bright=true ),
 " sample(s) parsed from ",
 address.hl( fgYellow, bright=true ),
 " in ", hl($( round(cpu_time() - t0, 3) ), fgWhite, bright=true), " seconds."
 # " ", hl($(round((get_occupied_mem()/1024/1024),1)), fgWhite, bright=true), "MB memory used."
 )
-proc serverloop: void =
+proc serverloop( conf: Config ): void =
 ## Open a database connection, bind to the listening socket,
 ## and start serving incoming netdata streams.
 let db = open( "", "", "", conf.dbopts )
-if conf.verbose: echo( "Successfully connected to the backend database.".hl( fgGreen ) )
+if conf.verbose: echo( "Successfully tested connection to the backend database.".hl( fgGreen ) )
+db.close
-var
-conn_count = 0
+# Setup listening socket.
-server = newSocket()
+#
+var server = newSocket()
 server.set_sock_opt( OptReuseAddr, true )
 server.bind_addr( Port(conf.listen_port), conf.listen_addr )
 server.listen()
 if conf.verbose:
 ":",
 hl( $conf.listen_port, fgBlue, bright=true ),
 )
 echo ""
+# Wait for incoming connections, fork for each client.
+#
 while true:
-var client  = newSocket()
+var
-var address = ""
+client  = new Socket
+address = ""
-# Force a garbage collection pass.
+status: cint = 0
-#
-conn_count = conn_count + 1
+server.acceptAddr( client, address ) # block
-if conn_count == 25:
-when defined( testing ): echo "Forcing GC pass."
+if fork() == 0:
-GC_full_collect()
+server.close
-conn_count = 1
+client.process( address )
+quit( 0 )
 client.close
-server.acceptAddr( client, address ) # blocking call
-spawn runthread( client, address, db, conf )
+discard waitpid( P_ALL, status, WNOHANG ) # reap all previous children
 when defined( testing ): dumpNumberOfInstances()
-proc parse_cmdline: void =
+proc parse_cmdline: Config =
 ## Populate the config object with the user's preferences.
+# Config object defaults.
+#
+result = Config(
+dbopts: "host=localhost port=5432 dbname=netdata user=netdata application_name=netdata-tsrelay",
+listen_port: 14866,
+listen_addr: "0.0.0.0",
+verbose: true,
+debug: false
+)
 # always set debug mode if development build.
-conf.debug = defined( testing )
+result.debug = defined( testing )
 for kind, key, val in getopt():
 case kind
 of cmdArgument:
 discard
 of cmdLongOption, cmdShortOption:
 case key
 of "debug", "d":
-conf.debug = true
+result.debug = true
-of "no-color", "c":
-conf.use_color = false
 of "help", "h":
 echo USAGE
 quit( 0 )
 of "quiet", "q":
-conf.verbose = false
+result.verbose = false
 of "version", "v":
 echo hl( "netdata_tsrelay " & VERSION, fgWhite, bright=true )
 quit( 0 )
-of "dbopts": conf.dbopts = val
+of "dbopts": result.dbopts = val
-of "listen-addr", "a": conf.listen_addr = val
+of "listen-addr", "a": result.listen_addr = val
-of "listen-port", "p": conf.listen_port = val.parse_int
+of "listen-port", "p": result.listen_port = val.parse_int
 else: discard
 of cmdEnd: assert( false ) # shouldn't reach here ever
 when isMainModule:
 system.addQuitProc( resetAttributes )
-parse_cmdline()
+conf = parse_cmdline()
 if conf.debug: echo hl( $conf, fgYellow )
-serverloop()
+serverloop( conf )

changeset 8	1ef3f2d6d10e
parent 7	c0bcf3bea772
child 9	aa9d537f7067