importing 14.5 release

Author: ivmfnal

DataProxy8.py

@@ -2,10 +2,10 @@
 #from Selector import Selector
 import yaml, socket, time, fnmatch, signal, select, sys, traceback
 from datetime import datetime, timedelta
-from LogFile import LogFile
+#from LogFile import LogFile
 from threading import RLock
 from WebInterface2 import DataProxyWebInterface
-from pythreader import TaskQueue, Task, PyThread, synchronized, Primitive, Scheduler
+from pythreader import TaskQueue, Task, PyThread, synchronized, Primitive, Scheduler, LogFile
 from HTTPProxy2 import HTTPProxy
 from DataLogger2 import DataLogger
 from record import TimeWindow
@@ -83,8 +83,8 @@ def run(self):
         for s in self.Servers.values():
             s.start()
         scheduler = Scheduler()
-        scheduler.add(self.tick, self.TickInterval)
-        scheduler.add(self.DataLogger.sendStatsToGraphite, 60.0, self)
+        scheduler.add(self.tick, interval=self.TickInterval)
+        #scheduler.add(self.DataLogger.sendStatsToGraphite, self, interval=60.0)
         scheduler.start()
         scheduler.join()
 
@@ -208,9 +208,17 @@ def run(self):
 
 if __name__ == "__main__":
     import traceback, sys, os, threading, getopt
+    from Version import Version
+    import webpie, pythreader
+    
+    print("Octopus:    ", Version, __file__)
+    print("webpie:     ", webpie.__version__, webpie.__file__)
+    print("pythreader: ", pythreader.__version__, pythreader.__file__)
+    print()
+    
     try:
         import logs
-        from webpie import HTTPServer, HTTPSServer
+        from webpie import HTTPServer
 
         open("DataProxy.pid", "w").write("%d" % (os.getpid(),))
 
@@ -277,23 +285,20 @@ def run(self):
             print("creating DataProxyWebInterface...")
             gui = DataProxyWebInterface(tm, title, data_logger, scanner_detector, static_location, correlations_file)
             port = cfg["port"]
-            tls = "tls" in cfg
             logging = log_file is not None
             debug = None        # for now
-            if tls:
-                keyfile = cfg["tls"]["key_file"]
-                certfile = cfg["tls"]["cert_file"]
-                ca_file = cfg["tls"].get("ca_file")
-                gui_server = HTTPSServer(port, gui, 
-                    certfile, keyfile, ca_file=ca_file,
-                    url_pattern = "*", max_connections = 20, max_queued = 10,
+
+            tls = cfg.get("tls", {})
+            keyfile = tls.get("key_file")
+            certfile = tls.get("cert_file")
+            ca_file = tls.get("ca_file")
+
+            gui_server = HTTPServer(port, gui, 
+                    certfile=certfile, keyfile=keyfile, ca_file=ca_file,
+                    max_connections = 20, max_queued = 10,
                     logging = logging, log_file=log_file, debug=debug)
-                gui_url = "https://%s:%s/index" % (socket.getfqdn(), port)
-            else:
-                gui_server = HTTPServer(port, gui, 
-                            url_pattern = "*", max_connections = 20, max_queued = 10,
-                            logging = logging, log_file=log_file, debug=debug)
-                gui_url = "http://%s:%s/index" % (socket.getfqdn(), port)
+            gui_http = "https" if tls else "http"
+            gui_url = f"{gui_http}://%s:%s/index" % (socket.getfqdn(), port)
             print("starting HTTPServer...")
             gui_server.start()
             gui_server.kind = "GUIServer"
@@ -302,7 +307,6 @@ def run(self):
                 rport = cfg["redirector"]["port"]
                 url = cfg["redirector"].get("url", gui_url)
                 gui_redirector = HTTPServer(rport, Redirector(url), 
-                        url_pattern = "*", 
                         max_connections = 20, max_queued = 10, logging = logging, log_file=log_file, debug=None)
                 gui_redirector.kind = "GUIRedirector"
                 print("starting Redirector...")

HTTPProxy2.py

@@ -21,7 +21,7 @@ def __init__(self, sock, my_name, header_received=False, nbytes=0):
     def read(self):
         try:    data = self.Sock.recv(self.MAXREAD)
         except socket.timeout:
-            self.debug("timeout")
+            #self.debug("timeout")
             data = b''
         except Exception as e: 
             self.debug("reader error: %s" % (e,))
@@ -269,9 +269,11 @@ def peerClosed(self, peer, exit_status, error_message=None):
                     sock.shutdown(socket.SHUT_WR)
                 except OSError as exc:
                     if exc.errno == errno.ENOTCONN:
-                        self.debug("not connected exception -- ignored")
+                        #self.debug("not connected exception -- ignored")
+                        pass
                     else:
-                        self.debug("Error shutting down write side: %s" % (traceback.format_exc(),))
+                        #self.debug("Error shutting down write side: %s" % (traceback.format_exc(),))
+                        pass
 
 
 if __name__ == "__main__":

VServer1.py

@@ -35,8 +35,7 @@ def dispatch(self, request):
             request.Dispatched = False
             request.DispatcherStatus = "nomatch"
             return None
-            
-            
+
 class RequestReaderTask(Task, Debugged):
 
     def __init__(self, request, vserver, dispatcher):
@@ -123,7 +122,7 @@ def run(self):
                 service = self.Dispatcher.dispatch(request)        
                 if not service:
                     error = "Service not found"
-                    http_status = 403
+                    http_status = 404
 
         if not request.Dispatched:
             request.Error = error
@@ -138,21 +137,6 @@ def run(self):
 
         return request
 
-class RequestReader(Debugged, Logged):
-    
-    def __init__(self, vserver, dispatcher, data_logger, config):
-        Debugged.__init__(self, "[RequestReader %s]" % (vserver.Port,))
-        Logged.__init__(self, "[RequestReader %s]" % (vserver.Port,))
-        self.VServer = vserver
-        self.Dispatcher = dispatcher
-        self.DataLogger = data_logger
-     
-    def put(self, request):
-        self.ReaderQueue.addTask(RequestReaderTask(request, self.VServer, self.Dispatcher))
-        
-    def counts(self):
-        return self.ReaderQueue.counts()
-                
 class VirtualServer(PyThread, Debugged, Logged):
 
     LoggerFields = [

Version.py

@@ -1 +1,5 @@
-Version="14.4"
+Version="14.5"
+
+if __name__ == "__main__":
+    print(Version)
+

debug.py

@@ -1,4 +1,4 @@
-from LogFile import LogFile, LogStream, make_timestamp
+from pythreader import LogFile, LogStream
 import sys
 from datetime import datetime
 
@@ -20,7 +20,15 @@ def debug(msg, t=None, add_timestamp=True):
         if add_timestamp or t is not None:
             msg = "%s: %s" % (make_timestamp(t), msg)
         _debug_out.log(msg)
-        
+
+def make_timestamp(t=None):
+    if t is None:   
+        t = datetime.datetime.now()
+    elif isinstance(t, (int, float)):
+        t = datetime.datetime.fromtimestamp(t)
+    return t.strftime("%m/%d/%Y %H:%M:%S") + ".%03d" % (t.microsecond//1000)
+
+
 class Debugged(object):
 
     def __init__(self, my_name=None, send_to=None, save_in_memory=False, add_timestamps=False, enabled=True):

logs.py

@@ -1,4 +1,4 @@
-from LogFile import LogFile, LogStream
+from pythreader import LogFile, LogStream
 from request_logger import RequestLogger
 import sys
 
@@ -12,20 +12,20 @@ def openLogFile(path):
     if path == "-":
         logfile = LogStream(sys.stdout)
     else:
-        logfile = LogFile(path)
+        logfile = LogFile(path, flush_interval=1.0)
         logfile.start()
 
 def openErrorFile(path):
     global errfile
     if path == "-":
         errfile = LogStream(sys.stdout)
     else:
-        errfile = LogFile(path)
+        errfile = LogFile(path, flush_interval=1.0)
         errfile.start()
 
 def openDebugFile(path):
     global debugfile
-    debugfile = LogStream(sys.stdout) if path == "-" else LogFile(path, interval="1h")
+    debugfile = LogStream(sys.stdout) if path == "-" else LogFile(path, interval="1h", flush_interval=1.0)
     debugfile.log("-------- Started --------")
 
 def openRequestLog(path, data_logger):

record.py

@@ -448,9 +448,9 @@ def _vacuum(self):
 
     def run(self):
         scheduler = Scheduler()
-        scheduler.add(self._flush,  self.SaveInterval)
-        scheduler.add(self._cleanup, 24*3600)
-        scheduler.add(self._vacuum, 24*3600)                  # once per day
+        scheduler.add(self._flush, interval=self.SaveInterval)
+        scheduler.add(self._cleanup, interval=24*3600)
+        scheduler.add(self._vacuum, interval=24*3600)                  # once per day
         scheduler.start()
         scheduler.join()
 

request.py

@@ -38,7 +38,6 @@ def __init__(self, csock, caddr, server_port):
         self.TransferStatus = None
 
         self.HTTPStatus = None
-        self.HTTPRequest = None
 
         self.VServerError = None
         self.ServiceName = None
@@ -78,4 +77,4 @@ def close(self, done):
             self.CSock = None
 
 
-        
+        

request_logger.py

@@ -1,4 +1,4 @@
-from LogFile import LogFile
+from pythreader import LogFile
 from request import Request
 from pythreader import Primitive
 from dns import DNS
@@ -7,7 +7,7 @@ class RequestLogger(Primitive):
 
     def __init__(self, logfile, data_logger):
         if isinstance(logfile, str):
-            logfile = LogFile(logfile)
+            logfile = LogFile(logfile, flush_interval=1.0)
 
         self.LogFile = logfile
         self.DataLogger = data_logger

service.py

@@ -34,12 +34,12 @@ def chooseServer(self, request, server_list):
             host, port = address
             if probe:
                     url = "http://%s:%s%s" % (host, port, probe)
-                    self.debug("Trying probe: %s, probe_timeout=%s" % (url,self.Service.ProbeTimeout))
+                    #self.debug("Trying probe: %s, probe_timeout=%s" % (url,self.Service.ProbeTimeout))
                     try:    urlopen(url, None, self.Service.ProbeTimeout)
                     except:
                         exctype, excvalue = sys.exc_info()[:2]
-                        self.debug("probe failed: %s, probe_timeout: %s, error:%s %s" % (
-                                url, self.Service.ProbeTimeout, exctype, excvalue))    
+                        #self.debug("probe failed: %s, probe_timeout: %s, error:%s %s" % (
+                        #        url, self.Service.ProbeTimeout, exctype, excvalue))    
                         server_list.mark_bad(server)
                         continue
 
@@ -58,7 +58,7 @@ def chooseServer(self, request, server_list):
             server_list.allocate(server)
             return s, server
         else:
-            self.debug("can not find available server")
+            #self.debug("can not find available server")
             return None, None
 
     def run(self):
@@ -145,6 +145,7 @@ def __init__(self, mgr, data_logger, name,
         self.RewriteURI = None  # see note about URL rewriting  # rewrite_uri   # tuple: ("old_head", "new_head") or None
         self.Manager = mgr
         self.MaxConnections = max_connections
+        self.MaxFrequency = max_frequency
         self.Backlog = backlog
 
         stagger = 1.0/max_frequency if max_frequency else None
@@ -159,8 +160,8 @@ def __init__(self, mgr, data_logger, name,
         self.ProbeTimeout = probe_timeout
         self.TransferTimeout = transfer_timeout
         if self.TransferTimeout != None:  self.TransferTimeout = float(self.TransferTimeout)
-        self.debug("Service created: %s, %s, %s, pt=%s, tt=%s, %s" % 
-                (name, max_connections, probe, probe_timeout, self.TransferTimeout, servers))
+        #self.debug("Service created: %s, %s, %s, pt=%s, tt=%s, %s" % 
+        #        (name, max_connections, probe, probe_timeout, self.TransferTimeout, servers))
         self.Access = access
         self.RemovePrefix = remove_prefix
         self.AddPrefix = add_prefix
@@ -404,7 +405,7 @@ def taskEnded(self, queue, task, request):
         request.TransferEndTime = task.Ended
         self.requestEnded(request)
         rid = request.Id
-        self.debug(f"{rid} transfer ended")
+        #self.debug(f"{rid} transfer ended")
 
     def taskFailed(self, queue, task, exc_type, exc_value, tb):
         request = task.Request

templates/services.html

@@ -9,11 +9,12 @@ <h2>Services</h2>
         <th rowspan=2>Service</th>
         <th rowspan=2>Server ports</th>
         <th rowspan=2>Match</th>
-        <th rowspan=2>Active Connections</th>
+        <th rowspan=2>Active connections</th>
         <th rowspan=2>Queue</th>
         <th colspan=3>Processing time</th>
         <th colspan=3>Wait time</th>
-        <th colspan=3>Request Counts</th>
+        <th rowspan=2>Max frequency</th>
+        <th colspan=3>Request counts</th>
         <th rowspan=2>Servers</th>
         <th rowspan=2>Probe</th>
         <th rowspan=2>Timeout</th>
@@ -37,6 +38,7 @@ <h2>Services</h2>
             <td class=left>{{s.currentQueueWaitTime()|pretty_time}}</td>
             <td class=none>{{s.AvgWaitT|pretty_time}}</td>
             <td class=right>{{s.MaxWaitT|pretty_time}}</td>
+            <td>{{s.MaxFrequency or ''}}</td>
             <td class=left>{{s.f1|pretty_frequency}}</td>
             <td class=none>{{s.f2|pretty_frequency}}</td>
             <td class=right>{{s.f3|pretty_frequency}}</td>