Checkmk git commits

checkmk-commits@lists.checkmk.com

1 participants
64454 discussions

Check_MK Git: check_mk: liveproxyd: first successful forwarding

by Mathias Kettner

Module: check_mk Branch: master Commit: 66a3f4148d9e74842323e300d7c3d5282c321873 URL: http://git.mathias-kettner.de/git/?p=check_mk.git;a=commit;h=66a3f4148d9e74… Author: Mathias Kettner <mk(a)mathias-kettner.de> Date: Tue Jun 4 11:41:25 2013 +0200 liveproxyd: first successful forwarding --- doc/treasures/liveproxy/liveproxyd | 130 +++++++++++++++++++++++++++++++----- 1 file changed, 114 insertions(+), 16 deletions(-) diff --git a/doc/treasures/liveproxy/liveproxyd b/doc/treasures/liveproxy/liveproxyd index d1313e4..3cb14ce 100755 --- a/doc/treasures/liveproxy/liveproxyd +++ b/doc/treasures/liveproxy/liveproxyd @@ -32,24 +32,26 @@ g_sites = { def liveproxyd_run(): open_client_sockets() - try: - while True: + while True: + try: initiate_connections() readable, writable = do_select() complete_connections(writable) accept_new_clients(readable) get_new_requests(readable) distribute_requests() - - - except MKSignalException, e: - log("Got signal %d." % e._signum) - if e._signum in [ 2, 3, 15 ]: - sys.exit(0) - elif e._signum == 10: - do_restart() - elif e._signum == 1: - do_reload() + get_responses(readable) + + except MKSignalException, e: + log("Got signal %d." % e._signum) + if e._signum in [ 2, 3, 15 ]: + sys.exit(0) + elif e._signum == 10: + do_restart() + elif e._signum == 12: + dump_state() + elif e._signum == 1: + do_reload() def initiate_connections(): # Create new channels to target sites. Nonblocking! @@ -104,6 +106,11 @@ def do_select(): if client["state"] == "idle": read_fds.append(client["socket"]) + # Responses from channels + for channel in sitestate["channels"]: + if channel["state"] == "busy": + read_fds.append(channel["socket"]) + r_able, w_able, x_able = select.select(read_fds, write_fds, [], 1.0) return r_able, w_able @@ -162,14 +169,16 @@ def forward_request(sitename, client, channel): chs.send(client["request"]) client["state"] = "wait_for_response" channel["state"] = "busy" + client["channel"] = channel + channel["client"] = client except Exception, e: if opt_debug: raise log("Error: %s" % e) - close_client_with_error(client, str(e)) + respond_client_with_error(client, str(e)) -def close_client_with_error(client, message): +def respond_client_with_error(client, message): try: client["socket"].send("400%12d\n%s\n" % (len(message) + 2, message)) except Exception, e: @@ -177,9 +186,15 @@ def close_client_with_error(client, message): raise log("Cannot send error message to client %s/%d: %s" % ( sitename, client["socket"].fileno(), e)) + client["state"] = "closed" + client["socket"] = None + + if "channel" in client: + channel = client["channel"] + del channel["client"] + del client["channel"] - client["socket"] = None - client["state"] = "closed" + client["state"] = "idle" @@ -192,6 +207,72 @@ def receive_request(sock): return textbody +def get_responses(readable): + for sitename, sitestate in g_sites.items(): + for channel in sitestate["channels"]: + if channel["state"] == "busy" and channel["socket"] in readable: + receive_response(sitename, channel) + + +def receive_response(sitename, channel): + client = channel["client"] + # We always assume fixed16 as response header! + old_response = channel.get("response", "") + try: + chunk = channel["socket"].recv(65536) + response = old_response + chunk + channel["response"] = response + except Exception, e: + if opt_debug: + raise + log("Cannot read response from %s/%d: %s" % + (sitename, channel["socket"].fileno(), e)) + channel["state"] = "error" + channel["socket"] = None + respond_client_with_error(client, "Error receiving response from target site %s: %s" % + (sitenam, e)) + + if len(response) < 16: # header not yet complete + return + + try: + bodylength = int(response[3:15]) + except Exception, e: + if opt_debug: + raise + log("Malformed response header from cannel %s/%d" % sitename, channel["socket"].fileno()) + respond_client_with_error(client, "Malformed response from site %s" % sitename) + channel["state"] = error + channel["socket"] = None + + if len(response) > bodylength + 16: + log("Too large response on channel %s/%d (%d exceeding bytes)" % ( + sitename, channel["socket"].fileno(), (len(response) - bodylength - 16))) + respond_client_with_error(client, "To long response from site %s" % sitename) + channel["state"] = error + channel["socket"] = None + + elif len(response) < bodylength + 16: + return + + # Response complete + channel["state"] = "ready" + del channel["client"] + del client["channel"] + + try: + client["socket"].send(response) + client["state"] = "idle" + except Exception, e: + if opt_debug: + raise + log("Cannot forward response from client %s/%d to client %s/%d: %s", + (sitename, channel["socket"].fileno(), sitename, client["socket"].fileno(), e)) + client["state"] = "error" + client["socket"] = None + + + def open_client_sockets(): if not os.path.exists(opt_socketdir): os.makedirs(opt_socketdir) @@ -216,6 +297,22 @@ def create_unix_socket(sitename): g_sites[sitename]["socket"] = s +def dump_state(): + log("Current connection state:") + for sitename, sitestate in g_sites.items(): + log("[%s]" % sitename) + log(" Channels:") + for channel in sitestate["channels"]: + log(" %s/%d - %s, client: %s" % + (sitename, channel["socket"].fileno(), channel["state"], channel.get("client"))) + log(" Clients:") + for client in sitestate["clients"]: + log(" %s/%d - %s, channel: %s" % + (sitename, client["socket"].fileno(), client["state"], client.get("channel"))) + + + + #. # .--Daemon/main---------------------------------------------------------. # | ____ __ _ | @@ -429,6 +526,7 @@ try: signal.signal(3, signal_handler) # QUIT signal.signal(15, signal_handler) # TERM signal.signal(10, signal_handler) # USR1 + signal.signal(12, signal_handler) # USR2 # Now let's go... liveproxyd_run()

11 years, 4 months

Check_MK Git: check_mk: liveproxyd: detect clients closing the socket

by Mathias Kettner

Module: check_mk Branch: master Commit: f560d385f372014efa93ecf0e7ca976f211120f6 URL: http://git.mathias-kettner.de/git/?p=check_mk.git;a=commit;h=f560d385f37201… Author: Mathias Kettner <mk(a)mathias-kettner.de> Date: Tue Jun 4 11:49:52 2013 +0200 liveproxyd: detect clients closing the socket --- doc/treasures/liveproxy/liveproxyd | 30 +++++++++++++++++++----------- 1 file changed, 19 insertions(+), 11 deletions(-) diff --git a/doc/treasures/liveproxy/liveproxyd b/doc/treasures/liveproxy/liveproxyd index 3cb14ce..3a68359 100755 --- a/doc/treasures/liveproxy/liveproxyd +++ b/doc/treasures/liveproxy/liveproxyd @@ -41,6 +41,7 @@ def liveproxyd_run(): get_new_requests(readable) distribute_requests() get_responses(readable) + # garbage_collect_sockets() except MKSignalException, e: log("Got signal %d." % e._signum) @@ -118,7 +119,6 @@ def do_select(): def accept_new_clients(readable): for sitename, sitestate in g_sites.items(): if sitestate["socket"] in readable: - print readable try: s, addrinfo = sitestate["socket"].accept() log("Accepted new client %s/%d" % (sitename, s.fileno())) @@ -134,12 +134,16 @@ def get_new_requests(readable): if client["state"] == "idle" and \ client["socket"] in readable: try: - request = receive_request(client["socket"]) - client["state"] = "wait_for_channel" - client["since"] = time.time() - client["request"] = request - log("Get new request from %s/%d (%d bytes)" % - (sitename, client["socket"].fileno(), len(request))) + request = receive_request(sitename, client["socket"]) + if request: + client["state"] = "wait_for_channel" + client["since"] = time.time() + client["request"] = request + log("Get new request from %s/%d (%d bytes)" % + (sitename, client["socket"].fileno(), len(request))) + else: + client["state"] = "closed" + client["socket"] = None except Exception, e: if opt_debug: raise @@ -200,10 +204,14 @@ def respond_client_with_error(client, message): # TODO: one malicious client can hang the whole proxy. In order # to prevent this we'd need partial requests... -def receive_request(sock): +def receive_request(sitename, sock): textbody = "" while not textbody.endswith("\n\n"): - textbody += sock.recv(65536) + chunk = sock.recv(65536) + if not chunk: + log("Client %s/%d closed connection." % (sitename, sock.fileno())) + return None + textbody += chunk return textbody @@ -304,11 +312,11 @@ def dump_state(): log(" Channels:") for channel in sitestate["channels"]: log(" %s/%d - %s, client: %s" % - (sitename, channel["socket"].fileno(), channel["state"], channel.get("client"))) + (sitename, channel["socket"].fileno(), channel["state"], channel.get("client") and channel["client"]["socket"].fileno() or "none")) log(" Clients:") for client in sitestate["clients"]: log(" %s/%d - %s, channel: %s" % - (sitename, client["socket"].fileno(), client["state"], client.get("channel"))) + (sitename, client["socket"].fileno(), client["state"], client.get("channel") and client["channel"]["socket"].fileno() or "none"))

11 years, 4 months

Check_MK Git: check_mk: First skeleton of livestatus proxy

by Mathias Kettner

Module: check_mk Branch: master Commit: a051c9cba3f0db90e93b966079883c88a669a370 URL: http://git.mathias-kettner.de/git/?p=check_mk.git;a=commit;h=a051c9cba3f0db… Author: Mathias Kettner <mk(a)mathias-kettner.de> Date: Tue Jun 4 10:46:24 2013 +0200 First skeleton of livestatus proxy --- doc/treasures/liveproxy/TODO | 11 + doc/treasures/liveproxy/liveproxyd | 370 +++++++++++++++++++++++++++++++++ doc/treasures/liveproxy/liveproxyd.mk | 8 + 3 files changed, 389 insertions(+) Diff: http://git.mathias-kettner.de/git/?p=check_mk.git;a=commitdiff;h=a051c9cba3…

11 years, 4 months

Check_MK Git: check_mk: Livestatus proxy: forwarding implemented

by Mathias Kettner

Module: check_mk Branch: master Commit: c0f162de2cab90ec10e882578fb90cc269fd32bc URL: http://git.mathias-kettner.de/git/?p=check_mk.git;a=commit;h=c0f162de2cab90… Author: Mathias Kettner <mk(a)mathias-kettner.de> Date: Tue Jun 4 11:14:38 2013 +0200 Livestatus proxy: forwarding implemented --- doc/treasures/liveproxy/liveproxyd | 107 +++++++++++++++++++++++++++++++----- 1 file changed, 92 insertions(+), 15 deletions(-) diff --git a/doc/treasures/liveproxy/liveproxyd b/doc/treasures/liveproxy/liveproxyd index f08771e..d1313e4 100755 --- a/doc/treasures/liveproxy/liveproxyd +++ b/doc/treasures/liveproxy/liveproxyd @@ -38,6 +38,8 @@ def liveproxyd_run(): readable, writable = do_select() complete_connections(writable) accept_new_clients(readable) + get_new_requests(readable) + distribute_requests() except MKSignalException, e: @@ -81,7 +83,7 @@ def complete_connections(writable): for channel in sitestate["channels"]: if channel["state"] == "connecting" and channel["socket"] in writable: log("Channel %s/%d successfully connected" % (sitename, channel["socket"].fileno())) - channel["state"] = "connected" + channel["state"] = "ready" # Master/Mega/Central select(). We are going to be the select() master. Harhar. def do_select(): @@ -97,10 +99,99 @@ def do_select(): # new client connections read_fds.append(sitestate["socket"]) + # new requests from existing clients + for client in sitestate["clients"]: + if client["state"] == "idle": + read_fds.append(client["socket"]) + r_able, w_able, x_able = select.select(read_fds, write_fds, [], 1.0) return r_able, w_able +def accept_new_clients(readable): + for sitename, sitestate in g_sites.items(): + if sitestate["socket"] in readable: + print readable + try: + s, addrinfo = sitestate["socket"].accept() + log("Accepted new client %s/%d" % (sitename, s.fileno())) + sitestate["clients"].append({"socket" : s, "state" : "idle", "since" : time.time()}) + except Exception, e: + if opt_debug: + raise + log("Failed to accept new client for %s: %s" % (sitename, e)) + +def get_new_requests(readable): + for sitename, sitestate in g_sites.items(): + for client in sitestate["clients"]: + if client["state"] == "idle" and \ + client["socket"] in readable: + try: + request = receive_request(client["socket"]) + client["state"] = "wait_for_channel" + client["since"] = time.time() + client["request"] = request + log("Get new request from %s/%d (%d bytes)" % + (sitename, client["socket"].fileno(), len(request))) + except Exception, e: + if opt_debug: + raise + log("Cannot read request from client %s/%d: %s" % + (sitename, client["socket"].fileno(), e)) + +def distribute_requests(): + for sitename, sitestate in g_sites.items(): + waiting_clients = [ client for client in sitestate["clients"] if client["state"] == "wait_for_channel"] + # Sort after waiting time, we should be fair to all... + waiting_clients.sort(cmp = lambda a, b: cmp(a["since"], b["since"])) + + for channel in sitestate["channels"]: + if not waiting_clients: + break + if channel["state"] == "ready": + client = waiting_clients[0] + del waiting_clients[0] + forward_request(sitename, client, channel) + +def forward_request(sitename, client, channel): + cls = client["socket"] + chs = channel["socket"] + log("Forwarding request from client %s/%d to channel %s/%d" % ( + sitename, cls.fileno(), sitename, chs.fileno())) + try: + chs.send(client["request"]) + client["state"] = "wait_for_response" + channel["state"] = "busy" + except Exception, e: + if opt_debug: + raise + log("Error: %s" % e) + close_client_with_error(client, str(e)) + + +def close_client_with_error(client, message): + try: + client["socket"].send("400%12d\n%s\n" % (len(message) + 2, message)) + except Exception, e: + if opt_debug: + raise + log("Cannot send error message to client %s/%d: %s" % ( + sitename, client["socket"].fileno(), e)) + + client["socket"] = None + client["state"] = "closed" + + + +# TODO: one malicious client can hang the whole proxy. In order +# to prevent this we'd need partial requests... +def receive_request(sock): + textbody = "" + while not textbody.endswith("\n\n"): + textbody += sock.recv(65536) + return textbody + + def open_client_sockets(): if not os.path.exists(opt_socketdir): os.makedirs(opt_socketdir) @@ -125,20 +216,6 @@ def create_unix_socket(sitename): g_sites[sitename]["socket"] = s -def accept_new_clients(readable): - for sitename, sitestate in g_sites.items(): - if sitestate["socket"] in readable: - print readable - try: - s, addrinfo = sitestate["socket"].accept() - log("Accepted new client %s/%d" % (sitename, s.fileno())) - sitestate["clients"].append(s) - except Exception, e: - if opt_debug: - raise - log("Failed to accept new client for %s: %s" % (sitename, e)) - - #. # .--Daemon/main---------------------------------------------------------. # | ____ __ _ |

11 years, 4 months

Check_MK Git: check_mk: Merge branch 'master' of ssh://mathias-kettner.de/ check_mk

by Mathias Kettner

Module: check_mk Branch: master Commit: fe8cc3aa434aebd7b34ff1ce046111fe2b1ccca5 URL: http://git.mathias-kettner.de/git/?p=check_mk.git;a=commit;h=fe8cc3aa434aeb… Author: Mathias Kettner <mk(a)mathias-kettner.de> Date: Tue Jun 4 17:43:50 2013 +0200 Merge branch 'master' of ssh://mathias-kettner.de/check_mk ---

11 years, 4 months

Check_MK Git: check_mk: liveproxy: fixed several side cases

by Mathias Kettner

Module: check_mk Branch: master Commit: ef958a73ac5d92c6b2aa96790b3b3618a9b00d33 URL: http://git.mathias-kettner.de/git/?p=check_mk.git;a=commit;h=ef958a73ac5d92… Author: Mathias Kettner <mk(a)mathias-kettner.de> Date: Tue Jun 4 15:29:39 2013 +0200 liveproxy: fixed several side cases --- doc/treasures/liveproxy/liveproxyd | 78 +++++++++++++++++++++--------------- 1 file changed, 45 insertions(+), 33 deletions(-) diff --git a/doc/treasures/liveproxy/liveproxyd b/doc/treasures/liveproxy/liveproxyd index 19df726..0e746c0 100755 --- a/doc/treasures/liveproxy/liveproxyd +++ b/doc/treasures/liveproxy/liveproxyd @@ -21,7 +21,7 @@ sites = {} # State of all sites g_sites = { # "mysite" : { -# "state" : "starting" / "up" +# "state" : "starting" / "up" / "busy" # "socket" : -> UNIX server socket, # "clients" : [ ... ], # "channels" : [ ... ], @@ -84,7 +84,6 @@ def initiate_connections(): if len(channels) < siteconf["channels"]: if time.time() - sitestate["last_failed_connect"] >= siteconf["retry"]: s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - log("Creating channel %s/%d..." % (sitename, s.fileno())) s.setblocking(0) try: s.connect(siteconf["socket"]) @@ -108,7 +107,7 @@ def initiate_connections(): def do_heartbeats(): now = time.time() for sitename, sitestate in g_sites.items(): - if sitestate["state"] == "ready": + if sitestate["state"] != "starting": rate, timeout = sites[sitename]["heartbeat"] channel = sitestate["heartbeat"]["channel"] since = sitestate["heartbeat"]["since"] @@ -133,7 +132,7 @@ def send_heartbeat(sitename, sitestate): while True: next_channel = None for channel in sitestate["channels"]: - if channel["state"] == "ready": + if channel["state"] != "starting": if next_channel == None or channel["since"] < next_channel["since"]: min_since = channel["since"] next_channel = channel @@ -143,7 +142,7 @@ def send_heartbeat(sitename, sitestate): else: channel = next_channel - log("Sending heartbeat to channel %s/%d" % (sitename, channel["socket"].fileno())) + # log("Sending heartbeat to channel %s/%d" % (sitename, channel["socket"].fileno())) try: channel["socket"].send("GET status\nKeepAlive: on\nColumns: execute_service_checks\n\n") channel["state"] = "heartbeat" @@ -233,9 +232,9 @@ def get_new_requests(readable): for sitename, sitestate in g_sites.items(): for client in sitestate["clients"]: if client["state"] == "idle" and \ - client["socket"] in readable: + (client["socket"] in readable or client.get("nextrequest")): try: - request = receive_request(sitename, client["socket"]) + request = receive_request(sitename, client) if request: client["state"] = "wait_for_channel" client["since"] = time.time() @@ -252,17 +251,24 @@ def get_new_requests(readable): def distribute_requests(): for sitename, sitestate in g_sites.items(): + if sitestate["state"] == "busy": + continue waiting_clients = [ client for client in sitestate["clients"] if client["state"] == "wait_for_channel"] # Sort after waiting time, we should be fair to all... waiting_clients.sort(cmp = lambda a, b: cmp(a["since"], b["since"])) - for channel in sitestate["channels"]: - if not waiting_clients: - break - if channel["state"] == "ready": - client = waiting_clients[0] - del waiting_clients[0] - forward_request(sitename, client, channel) + # one channel must always be kept for heartbeat + allowed_channels = len([c for c in sitestate["channels"] if c["state"] in ["ready", "heartbeat"]]) + if allowed_channels <= 1: + sitestate["state"] = "busy" + else: + for channel in sitestate["channels"]: + if not waiting_clients: + break + if channel["state"] == "ready": + client = waiting_clients[0] + del waiting_clients[0] + forward_request(sitename, client, channel) def forward_request(sitename, client, channel): cls = client["socket"] @@ -271,10 +277,14 @@ def forward_request(sitename, client, channel): client["request"], sitename, cls.fileno(), sitename, chs.fileno())) try: chs.send(client["request"]) - client["state"] = "wait_for_response" - channel["state"] = "busy" - client["channel"] = channel - channel["client"] = client + if not client["request"].startswith("COMMAND"): + client["state"] = "wait_for_response" + channel["state"] = "busy" + client["channel"] = channel + channel["client"] = client + else: + client["request"] = "" + client["state"] = "idle" except Exception, e: if opt_debug: raise @@ -304,20 +314,24 @@ def respond_client_with_error(client, message): # TODO: one malicious client can hang the whole proxy. In order # to prevent this we'd need partial requests... -def receive_request(sitename, sock): - textbody = "" - while not textbody.endswith("\n\n"): +def receive_request(sitename, client): + # Note: Multisite can send several requests at once. For example + # a command and a wait query (reschedule button) + request = client.get("nextrequest", "") + while "\n\n" not in request: try: - chunk = sock.recv(65536) - log("CHUNK: %s" % chunk) + chunk = client["socket"].recv(65536) + request += chunk except socket.error, e: if e.errno != 4: # Interrupted system cal raise if not chunk: - log("Client %s/%d closed connection." % (sitename, sock.fileno())) + log("Client %s/%d closed connection." % (sitename, client["socket"].fileno())) return None - textbody += chunk - return textbody + end = request.index("\n\n") + client["nextrequest"] = request[end+2:] + request = request[:end+2] + return request def get_responses(readable): @@ -332,14 +346,11 @@ def get_responses(readable): def receive_response(sitename, channel): - log("HIRN: Hole Antwort...") client = channel["client"] # We always assume fixed16 as response header! old_response = channel.get("response", "") - log("HIRN: altes test: %s" % old_response) try: chunk = channel["socket"].recv(65536) - log("HIRN: neuer Chunk: %s" % chunk) if not chunk: raise Exception("Connect closed by foreign host") @@ -355,12 +366,10 @@ def receive_response(sitename, channel): return if len(response) < 16: # header not yet complete - log("HIRN: noch nicht fertig, nur [%s]" % response) return try: bodylength = int(response[3:15]) - log("HIRN: koerper ist %d bytes" % bodylength) except Exception, e: if opt_debug: raise @@ -383,6 +392,7 @@ def receive_response(sitename, channel): # Response complete channel["state"] = "ready" + g_sites[sitename]["state"] = "ready" # at least one channel free channel["response"] = "" del channel["client"] del client["channel"] @@ -411,8 +421,8 @@ def receive_heartbeat(sitename, channel): sitename, channel["socket"].fileno(), chunk)) disconnect_from_site(sitename) else: - log("Channel %s/%d: received valid heartbeat" % ( - sitename, channel["socket"].fileno())) + # log("Channel %s/%d: received valid heartbeat" % ( + # sitename, channel["socket"].fileno())) channel["state"] = "ready" sitestate["heartbeat"]["since"] = time.time() sitestate["heartbeat"]["channel"] = None @@ -644,6 +654,8 @@ def read_configuration(): "last_reset" : time.time(), "last_failed_connect" : 0, } + if siteconf["channels"] <= 1: + raise bail_out("Invalid configuration for site %s: you need at least two channels" % sitename) def do_restart(): log("Restarting myself")

11 years, 4 months

Check_MK Git: check_mk: FIX: fix invalid request in livestatus query after reconnect

by Mathias Kettner

Module: check_mk Branch: master Commit: 841d4b05e1540975850a7905d3786d53f370b26d URL: http://git.mathias-kettner.de/git/?p=check_mk.git;a=commit;h=841d4b05e15409… Author: Mathias Kettner <mk(a)mathias-kettner.de> Date: Tue Jun 4 16:53:13 2013 +0200 FIX: fix invalid request in livestatus query after reconnect --- ChangeLog | 1 + web/htdocs/livestatus.py | 3 ++- 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/ChangeLog b/ChangeLog index 1daa238..2c27ccc 100644 --- a/ChangeLog +++ b/ChangeLog @@ -185,6 +185,7 @@ Multisite: * List of views: Output the alias of a datasource instead of internal name * FIX: fix column editor for join columns if "SERVICE:" is l10n'ed + * FIX: fix invalid request in livestatus query after reconnect WATO: * FIX: convert editing of global setting to POST. This avoid URL-too-long diff --git a/web/htdocs/livestatus.py b/web/htdocs/livestatus.py index bff1cfd..640835a 100644 --- a/web/htdocs/livestatus.py +++ b/web/htdocs/livestatus.py @@ -261,6 +261,7 @@ class BaseConnection: return self.recv_response(query, add_headers) def send_query(self, query, add_headers = "", do_reconnect=True): + orig_query = query if self.socket == None: self.connect() if not query.endswith("\n"): @@ -288,7 +289,7 @@ class BaseConnection: # Automatically try to reconnect in case of an error, but # only once. self.connect() - self.send_query(query, add_headers, False) + self.send_query(orig_query, add_headers, False) return raise MKLivestatusSocketError("RC1:" + str(e))

11 years, 4 months

Check_MK Git: check_mk: liveproxyd: removed some debug messages

by Mathias Kettner

Module: check_mk Branch: master Commit: da8fd7fd0d922c0e1fdbe8d9467a11eaadb588ae URL: http://git.mathias-kettner.de/git/?p=check_mk.git;a=commit;h=da8fd7fd0d922c… Author: Mathias Kettner <mk(a)mathias-kettner.de> Date: Tue Jun 4 16:55:57 2013 +0200 liveproxyd: removed some debug messages --- doc/treasures/liveproxy/liveproxyd | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/doc/treasures/liveproxy/liveproxyd b/doc/treasures/liveproxy/liveproxyd index c64224d..2bb98a3 100755 --- a/doc/treasures/liveproxy/liveproxyd +++ b/doc/treasures/liveproxy/liveproxyd @@ -247,8 +247,8 @@ def get_new_requests(readable): client["state"] = "wait_for_channel" client["since"] = time.time() client["request"] = request - log("Get new request from %s/%d (%d bytes)" % - (sitename, client["socket"].fileno(), len(request))) + # log("Get new request from %s/%d (%d bytes)" % + # (sitename, client["socket"].fileno(), len(request))) else: client["state"] = "closed" except Exception, e: @@ -281,8 +281,8 @@ def distribute_requests(): def forward_request(sitename, client, channel): cls = client["socket"] chs = channel["socket"] - log("Forwarding request [%s] from client %s/%d to channel %s/%d" % ( - client["request"].replace("\n","\\n"), sitename, cls.fileno(), sitename, chs.fileno())) + # log("Forwarding request [%s] from client %s/%d to channel %s/%d" % ( + # client["request"].replace("\n","\\n"), sitename, cls.fileno(), sitename, chs.fileno())) try: chs.send(client["request"]) if not client["request"].startswith("COMMAND"): @@ -333,14 +333,14 @@ def receive_request(sitename, client): # Note: Multisite can send several requests at once. For example # a command and a wait query (reschedule button) request = client.get("nextrequest", "") - if request: - log("HIRN: client %s/%d noch ein Rest vom letzten mal: [%s]" % ( - sitename, client["socket"].fileno(), request.replace("\n", "\\n"))) + # if request: + # log("HIRN: client %s/%d noch ein Rest vom letzten mal: [%s]" % ( + # sitename, client["socket"].fileno(), request.replace("\n", "\\n"))) client["nextrequest"] = "" while "\n\n" not in request: try: chunk = client["socket"].recv(65536) - log("HIRN: CHUNK von client: [%s]" % chunk.replace("\n", "\\n")) + # log("HIRN: CHUNK von client: [%s]" % chunk.replace("\n", "\\n")) request += chunk except socket.error, e: if e.errno == 104: # Connection reset by peer @@ -351,12 +351,12 @@ def receive_request(sitename, client): if not chunk: log("Client %s/%d closed connection." % (sitename, client["socket"].fileno())) return None - log("HIRN: VOR DEM SPLITTEN: [%s]" % request) + # log("HIRN: VOR DEM SPLITTEN: [%s]" % request) end = request.index("\n\n") client["nextrequest"] = request[end+2:] request = request[:end+2] - log("HIRN: Diesesmal: [%s], naechstes Mal: [%s]" % ( - request.replace("\n", "\\n"), client["nextrequest"].replace("\n", "\\n"))) + #log("HIRN: Diesesmal: [%s], naechstes Mal: [%s]" % ( + # request.replace("\n", "\\n"), client["nextrequest"].replace("\n", "\\n"))) return request

11 years, 4 months

Check_MK Git: check_mk: Multisite: enable liveproxyd

by Mathias Kettner

Module: check_mk Branch: master Commit: 8d44d13e76c0f9395768104bc3614a70362005fa URL: http://git.mathias-kettner.de/git/?p=check_mk.git;a=commit;h=8d44d13e76c0f9… Author: Mathias Kettner <mk(a)mathias-kettner.de> Date: Tue Jun 4 17:43:46 2013 +0200 Multisite: enable liveproxyd --- doc/treasures/liveproxy/liveproxyd | 7 +- web/htdocs/index.py | 5 ++ web/htdocs/wato.py | 135 +++++++++++++++++++++++++++++------- web/plugins/config/builtin.py | 3 + 4 files changed, 123 insertions(+), 27 deletions(-) diff --git a/doc/treasures/liveproxy/liveproxyd b/doc/treasures/liveproxy/liveproxyd index 2bb98a3..b77d0b7 100755 --- a/doc/treasures/liveproxy/liveproxyd +++ b/doc/treasures/liveproxy/liveproxyd @@ -86,7 +86,7 @@ def initiate_connections(): sitestate = g_sites[sitename] channels = sitestate["channels"] if len(channels) < siteconf["channels"]: - if time.time() - sitestate["last_failed_connect"] >= siteconf["retry"]: + if time.time() - sitestate["last_failed_connect"] >= siteconf["connect_retry"]: s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.setblocking(0) try: @@ -693,6 +693,11 @@ def read_configuration(): # Create state object for all sites for sitename, siteconf in sites.items(): + siteconf.setdefault("channels", 5) + siteconf.setdefault("heartbeat", (5, 2.0)) + siteconf.setdefault("channel_timeout", 3.0) + siteconf.setdefault("query_timeout", 120.0) + siteconf.setdefault("connect_retry", 4.0) g_sites[sitename] = { "state": "starting", "channels" : [], diff --git a/web/htdocs/index.py b/web/htdocs/index.py index 763f3c3..f70d908 100644 --- a/web/htdocs/index.py +++ b/web/htdocs/index.py @@ -112,6 +112,11 @@ def connect_to_livestatus(html): for sitename, site in config.allsites().items(): siteconf = config.user_siteconf.get(sitename, {}) + # Convert livestatus-proxy links into UNIX socket + s = site["socket"] + if type(s) == tuple and s[0] == "proxy": + site["socket"] = "unix:" + defaults.livestatus_unix_socket + "proxy/" + sitename + if siteconf.get("disabled", False): html.site_status[sitename] = { "state" : "disabled", "site" : site } disabled_sites[sitename] = site diff --git a/web/htdocs/wato.py b/web/htdocs/wato.py index 64f3961..6aed65c 100644 --- a/web/htdocs/wato.py +++ b/web/htdocs/wato.py @@ -3780,14 +3780,15 @@ def log_pending(status, linkinfo, what, message): def cmc_rush_ahead(): if defaults.omd_root: socket_path = defaults.omd_root + "/tmp/run/cmcrush" - try: - changeid = str(random.randint(1, 100000000000000000)) - file(log_dir + "changeid", "w").write(changeid + "\n") - socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) \ - .sendto(changeid, socket_path) - except: - if config.debug: - raise + if os.path.exists(socket_path): + try: + changeid = str(random.randint(1, 100000000000000000)) + file(log_dir + "changeid", "w").write(changeid + "\n") + socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) \ + .sendto(changeid, socket_path) + except: + if config.debug: + raise def log_commit_pending(): @@ -6531,7 +6532,11 @@ def mode_sites(phase): socket = site.get("socket", _("local site")) if socket == "disabled:": socket = _("don't query status") - table.cell(_("Socket"), socket) + table.cell(_("Socket")) + if type(socket) == tuple and socket[0] == "proxy": + html.write(_("Use livestatus Proxy-Daemon")) + else: + html.write(socket) # Status host if "status_host" in site: @@ -6602,23 +6607,77 @@ def mode_edit_site(phase): else: site = sites.get(siteid, {}) + vs_tcp_port = Tuple( + title = _("TCP Port to connect to"), + orientation = "float", + elements = [ + TextAscii(label = _("Host:"), allow_empty = False, size=15), + Integer(label = _("Port:"), minvalue=1, maxvalue=65535, default_value=6557), + ]) + conn_choices = [ + ( None, _("Connect to the local site") ), + ( "tcp", _("Connect via TCP"), vs_tcp_port), + ( "unix", _("Connect via UNIX socket"), TextAscii( + label = _("Path:"), + size = 40, + allow_empty = False)), + ( "disabled", _("Do not connect")), + ] + if config.liveproxyd_enabled: + conn_choices[2:2] = [ + ( "proxy", _("Use Livestatus Proxy-Daemon"), + Dictionary( + optional_keys = False, + columns = 1, + elements = [ + ( "socket", vs_tcp_port ), + ( "channels", + Integer( + title = _("Number of channels to keep open"), + minvalue = 2, + maxvalue = 50, + default_value = 5)), + ( "heartbeat", + Tuple( + title = _("Regular heartbeat"), + orientation = "float", + elements = [ + Integer(label = _("Rate:"), unit=_("/sec"), minvalue=1, default_value = 5), + Float(label = _("Timeout:"), unit=_("sec"), minvalue=0.1, default_value = 2.0), + ])), + ( "channel_timeout", + Float( + title = _("Timeout waiting for a free channel"), + minvalue = 0.1, + default_value = 3, + unit = _("sec"), + ) + ), + ( "query_timeout", + Float( + title = _("Total query timeout"), + minvalue = 0.1, + unit = _("sec"), + default_value = 120, + ) + ), + ( "connect_retry", + Float( + title = _("Wait time after failed connect"), + minvalue = 0.1, + unit = _("sec"), + default_value = 4.0, + )), + ] + ) + ) + ] + # ValueSpecs for the more complex input fields vs_conn_method = CascadingDropdown( html_separator = " ", - choices = [ - ( None, _("Connect to the local site") ), - ( "tcp", _("Connect via TCP"), Tuple( - orientation = "float", - elements = [ - TextAscii(label = _("Host:"), allow_empty = False, size=15), - Integer(label = _("Port:"), minvalue=1, maxvalue=65535, default_value=6557), - ])), - ( "unix", _("Connect via UNIX socket"), TextAscii( - label = _("Path:"), - size = 30, - allow_empty = False)), - ( "disabled", _("Do not connect")), - ]) + choices = conn_choices, + ) if phase == "action": @@ -6657,7 +6716,7 @@ def mode_edit_site(phase): # Connection method = vs_conn_method.from_html_vars("method") vs_conn_method.validate_value(method, "method") - if type(method) == tuple: + if type(method) == tuple and method[0] in [ "unix", "tcp"]: if method[0] == "unix": new_site["socket"] = "unix:" + method[1] else: @@ -6776,9 +6835,9 @@ def mode_edit_site(phase): forms.header(_("Livestatus settings")) forms.section(_("Connection")) method = site.get("socket", None) - if method and method.startswith("unix:"): + if type(method) == str and method.startswith("unix:"): method = ('unix', method[5:]) - elif method and method.startswith("tcp:"): + elif type(method) == str and method.startswith("tcp:"): method = ('tcp', tuple(method.split(":")[1:])) vs_conn_method.render_input("method", method) @@ -6904,6 +6963,7 @@ def load_sites(): def save_sites(sites): make_nagios_directory(multisite_dir) + # Important: even write out sites if it's empty. The global 'sites' # variable will otherwise survive in the Python interpreter of the # Apache processes. @@ -6916,9 +6976,32 @@ def save_sites(sites): rewrite_config_files_below(g_root_folder) # fix site attributes need_sidebar_reload() + if config.liveproxyd_enabled: + save_liveproxyd_config(sites) + # Call the sites saved hook call_hook_sites_saved(sites) +def save_liveproxyd_config(sites): + path = defaults.default_config_dir + "/liveproxyd.mk" + out = create_user_file(path, "w") + out.write("# Written by WATO\n# encoding: utf-8\n\n") + + conf = {} + for siteid, siteconf in sites.items(): + s = siteconf.get("socket") + if type(s) == tuple and s[0] == "proxy": + conf[siteid] = s[1] + + out.write("sites = \\\n%s\n" % pprint.pformat(conf)) + try: + pidfile = defaults.livestatus_unix_socket + "proxyd.pid" + pid = int(file(pidfile).read().strip()) + os.kill(pid, 10) + except Exception, e: + html.show_error(_("Warning: cannot reload Livestatus Proxy-Daemon: %s" % e)) + + # Makes sure, that in distributed mode we monitor only # the hosts that are directly assigned to our (the local) # site. diff --git a/web/plugins/config/builtin.py b/web/plugins/config/builtin.py index 8011c12..0cfdaa6 100644 --- a/web/plugins/config/builtin.py +++ b/web/plugins/config/builtin.py @@ -123,6 +123,9 @@ debug_livestatus_queries = False # not reachable. show_livestatus_errors = True +# Whether the livestatu proxy daemon is available +liveproxyd_enabled = False + # Set this to a list in order to globally control which views are # being displayed in the sidebar snapin "Views" visible_views = None

11 years, 4 months

Check_MK Git: check_mk: liveproxyd: heartbeat handling works

by Mathias Kettner

Module: check_mk Branch: master Commit: 535f95031580bbc80dbe5316c756da2f52601ec0 URL: http://git.mathias-kettner.de/git/?p=check_mk.git;a=commit;h=535f95031580bb… Author: Mathias Kettner <mk(a)mathias-kettner.de> Date: Tue Jun 4 15:04:01 2013 +0200 liveproxyd: heartbeat handling works --- doc/treasures/liveproxy/liveproxyd | 315 ++++++++++++++++++++++++++++-------- 1 file changed, 249 insertions(+), 66 deletions(-) Diff: http://git.mathias-kettner.de/git/?p=check_mk.git;a=commitdiff;h=535f950315…

11 years, 4 months

Jump to page:

2024

2023

2022

2021

2020

2019

2018

2017

2016

2015

2014

2013

2012

2011

2010

Checkmk git commits