From 30a0b3c1b5e9dfe7bd68a001f174cb5ab470ba1c Mon Sep 17 00:00:00 2001
From: Matt Rendina <mrendina@stsci.edu>
Date: Wed, 28 Aug 2019 13:06:08 -0400
Subject: Several updates. Added hashing support of input files. Generalized
 logfile regex. Simplified filtering of packages without user agent.
 Additional fraction metrics output.

---
 logparse.py | 248 ++++++++++++++++++++++++++++++++++++++++++------------------
 1 file changed, 173 insertions(+), 75 deletions(-)

diff --git a/logparse.py b/logparse.py
index 4b8edb0..baee14d 100755
--- a/logparse.py
+++ b/logparse.py
@@ -6,6 +6,7 @@ import re
 from glob import glob
 import argparse
 from math import ceil
+import hashlib
 import gzip
 import socket
 import pandas as pd
@@ -14,39 +15,58 @@ import matplotlib.pyplot as plt
 import matplotlib.dates as mdates
 from dateutil import parser as dpar
 from collections import OrderedDict
+import yaml
+
+
+def md5(fname):
+    hash_md5 = hashlib.md5()
+    with open(fname, "rb") as f:
+        for chunk in iter(lambda: f.read(4096), b""):
+            hash_md5.update(chunk)
+    return hash_md5.hexdigest()
+
 
 # regex pattern to extract key values from each line of an apache/nginx access log
 # Accommodate PUTs as well as second URLs (normally "-")
-patt = '(?P<ipaddress>\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}) .* .* \\[(?P<date>\\d{2}\\/[a-zA-Z]{3}\\/\\d{4}):(?P<time>\\d{2}:\\d{2}:\\d{2}) (\\+|\\-)\\d{4}] ".* (?P<path>.*?) .*" (?P<status>\\d*) \\d* ".*" "(?P<agent>.*)"'
+patt = '(?P<ipaddress>\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}) .* .* \\[(?P<date>\\d{2}\\/[a-zA-Z]{3}\\/\\d{4}):(?P<time>\\d{2}:\\d{2}:\\d{2}) (\\+|\\-)\\d{4}] ".* (?P<path>.*?) .*" (?P<status>\\d*) (?P<size>\\d*)'
         
 p = re.compile(patt)
 
-
 class logData():
 
+    columns = {
+        'ipaddress': {},
+        'hostname': {},
+        'date': {},
+        'time': {},
+        'path': {},
+        'status': {},
+        'agent': {},
+    }
+
     def __init__(self,
                  gethostnames=False,
                  ignore_hosts=[]):
-        self.columns = {
-                'ipaddress': {},
-                'hostname': {},
-                'date': {},
-                'time': {},
-                'path': {},
-                'status': {},
-                'agent': {},
-               }
-        self.dframe = pd.DataFrame(self.columns)
         self.digest_path = 'digests'
         self.gethostnames = gethostnames
+        self.hostnames = {}
         self.ignore_hosts = ignore_hosts
 
+    def poll_hostnames(self):
+        if ipaddress not in self.hostnames.keys():
+            try:
+                hostname = socket.gethostbyaddr(ipaddress)
+            except:
+                hostname = 'offline'
+            self.hostnames[ipaddress] = hostname
+        else:
+            hostname = self.hostnames[ipaddress]
+
     def process_lines(self, f):
         print('process lines')
         df = pd.DataFrame(self.columns)
         unparseable = 0
         for line in f.readlines():
-            print(line)
             try:
                 line = str(line.decode("utf-8"))
             except(AttributeError):
@@ -57,13 +77,13 @@ class logData():
                     if host in line:
                         continue
             except(TypeError):
-                continue
+                pass
             try:
                 match = p.match(line)
             except:
                 line_errors += 1
-                pass
-            print(match)
+                print(f'Line parse error: {line}')
+                continue
             try:
                 ipaddress = match.group('ipaddress')
                 date = match.group('date')
@@ -71,18 +91,18 @@ class logData():
                 time = match.group('time')
                 path = match.group('path')
                 status = match.group('status')
-                agent = match.group('agent')
+                #agent = match.group('agent')
+                hostname = ''
+                df = df.append({'ipaddress':ipaddress,
+                                'hostname':hostname,
+                                'date':dateobj,
+                                'time':time,
+                                'path':path,
+                                'status':status},
+                                ignore_index=True)
+                                #'agent':agent}, ignore_index=True)
             except(AttributeError):
                 unparseable += 1
-            # Selective polling of hostnames here.
-            hostname = '?'
-            df = df.append({'ipaddress':ipaddress,
-                            'hostname':hostname,
-                            'date':dateobj,
-                            'time':time,
-                            'path':path, 
-                            'status':status, 
-                            'agent':agent}, ignore_index=True)
         print(f'unparseable lines : {unparseable}')
         return(df)
     
@@ -90,35 +110,44 @@ class logData():
         '''Accepts:
     
         a list of apache/nginx access log files, either raw or .gz,
-        and parses each that does not already have a corresponding digested
-        data frame in the 'digests' subdir.'''
+        and parses each that has not already been ingested.'''
     
-        # Create data frame for receiving log data
-        df = pd.DataFrame(self.columns)
+        # Create data frame for receiving all log data.
         locframe = pd.DataFrame(self.columns)
-    
-        # Sort list of logs before processing so data will be appended in
-        # chronological order.
+
+        # Track which files have been parsed by storing the MD5 hash of each
+        # once it's been read.
+        pfile = 'parsed_files.dat'
+        if not os.path.exists(pfile):
+            open(pfile, 'a').close()
+        with open(pfile, 'r') as f:
+            already_parsed = f.read().split()
+        parsed = open(pfile, 'a')
+
         for log in sorted(logs):
+            # Compute MD5 hash of file and compare to list of files that
+            # have already been parsed. If new, parse, if not, skip.
+            hashval = md5(log)
+            if hashval in already_parsed:
+                print(f'File {log} already parsed.')
+                continue
+            df = pd.DataFrame(self.columns)
             setname = re.sub('\.gz$', '', log)
             setpath = os.path.join(self.digest_path, setname)
             pklpath = os.path.join(self.digest_path, f'{setname}.pkl')
             print(f'ingesting dataset = {setname}')
-            if os.path.isfile(pklpath):
-                df = pd.read_pickle(pklpath)
+            if '.gz' in log:
+                with gzip.open(log, 'r') as f:
+                    df = self.process_lines(f)
             else:
-                print('parsing log file')
-                if '.gz' in log:
-                    with gzip.open(log, 'r') as f:
-                        df = self.process_lines(f)
-                else:
-                    with open(log, 'r') as f:
-                        df = self.process_lines(f)
-                print(f'df shape = {df.shape}')
-            # Dump digested log data to disk for more efficient repeated use.
-            df.to_pickle(f'{setpath}.pkl')
+                with open(log, 'r') as f:
+                    df = self.process_lines(f)
+            print(f'df shape = {df.shape}')
             locframe = locframe.append(df, ignore_index=True)
             print(locframe.shape)
+            parsed.write(f'{hashval}\n')
+
+        parsed.close()
         return(locframe)
 
 
@@ -126,23 +155,28 @@ class logData():
 def filter_pkgs(df):
     '''Filter dataframe df down to just the rows the represent
     successful (HTTP 200) conda package (.bz2 files) downloads.'''
+    print(df)
     inlen = len(df)
-    out = df.loc[df['agent'].str.contains('conda')]
-    print(out)
-    out = out.loc[out['path'].str.contains('bz2')]
-    out = out.loc[out['status'].str.contains('200')]
+    ##out = df.loc[df['agent'].str.contains('conda')]
+    ##out = out.loc[out['path'].str.contains('bz2')]
+    out = df.loc[df['path'].str.contains('bz2')]
+    out = out.loc[(out['status'] == '200') | (out['status'] == '302')]
     outlen = len(out)
     print(f'{inlen-outlen} rows removed to leave conda txns only')
     return(out)
 
 
 
-
 def main():
     ap = argparse.ArgumentParser(
             prog='logparse.py',
             description='Parse and digest apache/nginx access logs in either'
             ' raw or .gz format.')
+    ap.add_argument('--config',
+                    '-c',
+                    help='Configuration file used to adjust behavior of the '
+                    'program',
+                    required=True)
     ap.add_argument('--files',
                     '-f',
                     help='List of log files to parse, raw or .gz are accepted.'
@@ -156,6 +190,9 @@ def main():
                     nargs='+')
     args = ap.parse_args()
 
+    with open(args.config, 'r') as f:
+        config = yaml.safe_load(f)
+
     files = []
     for filespec in args.files:
         expanded =  glob(filespec)
@@ -166,38 +203,66 @@ def main():
         else:
             files.append(expanded)
 
+    inf_hosts = config['infrastructure_hosts']
+    num_inf_hosts = len(inf_hosts)
+
+    # Read in any pre-existing parsed data.
+    datfile = 'dataframe.dat'
+    if not os.path.exists(datfile):
+        open(datfile, 'a').close()
+
+    with open(datfile, 'r') as f:
+        try:
+            data = pd.read_pickle(datfile)
+        except:
+            data = pd.DataFrame(logData.columns)
+
     # TODO: Should host filtering take place here?
-    #       It leaves a disconnect between the pickled data which _may_ have been
-    #       culled and the actual data being referenced by the inclusion of a file
-    #       that has data from an exluded host within it.
+    #       It leaves a disconnect between the pickled data which _may_ have
+    #       been culled and the actual data being referenced.
     logproc = logData(ignore_hosts=args.ignorehosts)
-    data = logproc.read_logs(files)
+    newdata = logproc.read_logs(files)
 
-    allpkgs = filter_pkgs(data)
-    allpkgs = allpkgs.sort_values(by='date')
+    # If any new log files were read, filter down to only conda package downloads
+    # Then sort by date.
+    if len(newdata.index) != 0:
+        newdata = filter_pkgs(newdata)
+        newdata = newdata.sort_values(by='date')
+
+    # Append newdata to existing data (potentially empty)
+    data = data.append(newdata, ignore_index=True)
+
+    # Remove any duplicate rows in data:
+    data = data.drop_duplicates()
+
+    # Dump data to disk for use during subsequent runs.
+    data.to_pickle(datfile)
 
-    start_date = allpkgs.iloc[0]['date']
-    end_date = allpkgs.iloc[-1]['date']
-    time_range = end_date - start_date
-    days_elapsed = time_range.days
-    if days_elapsed == 0:
-        days_elapsed = 1
-    
-    print(f'Over the period {start_date.strftime("%m-%d-%Y")} '
-          f'to {end_date.strftime("%m-%d-%Y")}')
-    print(f'{days_elapsed} days')
 
     # Normalize all conda-dev channel names to astroconda-dev
-    allpkgs = allpkgs.replace('/conda-dev', '/astroconda-dev', regex=True)
+    data = data.replace('/conda-dev', '/astroconda-dev', regex=True)
+
+    all_unique_hosts = list(set(data['ipaddress']))
+    #for host in all_unique_hosts:
+    #    try:
+    #        print(f'{host} {socket.gethostbyaddr(host)[0]}')
+    #    except:
+    #        print(f'{host} offline?')
 
     # All packages in a dictionary by channel.
-    chans = [path.split('/')[1] for path in allpkgs['path']]
-    chans = set(chans)
-    chan_pkgs = {}
+    chans = [path.split('/')[1] for path in data['path']]
+    chans = list(set(chans))
+    chans.sort()
+    chan_pkgs = OrderedDict()
     for chan in chans:
         # Trailing '/' added to ensure only a single channel gets stored for each
         # due to matching overlap depending on length of substring.
-        chan_pkgs[chan] = allpkgs[allpkgs['path'].str.contains(chan+'/')]
+        chan_pkgs[chan] = data[data['path'].str.contains(chan+'/')]
+
+    total_downloads = 0
+    for chan in chan_pkgs.keys():
+        total_downloads += len(chan_pkgs[chan].index)
+    print(f'TOTAL downloads = {total_downloads}')
 
     # For each channel, generate summary report of the download activity.
     for chan in chan_pkgs.keys():
@@ -211,21 +276,32 @@ def main():
         dates.sort()
         bydate = OrderedDict()
 
+        start_date = dates[0]
+        end_date = dates[-1]
+        time_range = end_date - start_date
+        days_elapsed = time_range.days
+        if days_elapsed == 0:
+            days_elapsed = 1
+        print(f'\nOver the period {start_date.strftime("%m-%d-%Y")} '
+              f'to {end_date.strftime("%m-%d-%Y")}')
+        print(f'{days_elapsed} days')
+
         # Downloads per day over time frame
         for date in dates:
             bydate[date] = len(pkgs[pkgs['date'] == date])
         #for date in bydate:
         #    print(f'{date} : {bydate[date]}')
 
-        total_downloads = len(pkgs.index)
-        print(f'Total downloads: {total_downloads}')
+        chan_downloads = len(pkgs.index)
+        print(f'Downloads: {chan_downloads}')
         # Downloads per week over time frame
 
-        print(f'Average downloads per day: {ceil(total_downloads / days_elapsed)}')
+        print(f'Average downloads per day: {ceil(chan_downloads / days_elapsed)}')
 
         # Number of unique hosts and geographic location
         unique_hosts = set(pkgs['ipaddress'])
-        print(f'Unique hosts {len(unique_hosts)}')
+        num_unique_hosts = len(unique_hosts)
+        print(f'Unique hosts {num_unique_hosts}')
 
         ## Unique packages
         unique_pkgs = set(pkgs['path'])
@@ -245,6 +321,23 @@ def main():
         #for i in range(top):
         #    print(pkg_totals[i])
 
+        # What fraction of total downloads come from non-infrastructure on-site hosts?
+        noninf = pkgs[~pkgs['ipaddress'].isin(config['infrastructure_hosts'])]
+        total_noninf = len(noninf.index)
+        print(f'Non-infrastructure downloads: {total_noninf}')
+        print(f'Percentage noninf downloads: {(total_noninf/chan_downloads)*100:.1f}%')
+
+        # What fraction of total downloads come from off-site hosts?
+        int_host_patterns = ['^'+s for s in config['internal_host_specs']]
+        offsite = pkgs[~pkgs['ipaddress'].str.contains(
+            '|'.join(int_host_patterns), regex=True)]
+        num_offsite_hosts = len(set(offsite['ipaddress']))
+        print(f'num off-site hosts: {num_offsite_hosts}')
+        onsite = pkgs[pkgs['ipaddress'].str.contains(
+            '|'.join(int_host_patterns), regex=True)]
+        num_onsite_hosts = len(set(onsite['ipaddress']))
+        print(f'num on-site hosts: {num_onsite_hosts}')
+
         # Totals of unique software names
         # i.e. name without version, hash, py or build iteration values
         # Extract simple package titles from 'path' column of data frame.
@@ -260,8 +353,13 @@ def main():
             total = names.count(name)
             name_totals.append([name, total])
         name_totals.sort(key=lambda x: x[1], reverse=True)
+        y = []
+        x = range(0,len(name_totals))
         for total in name_totals:
+            y.append(total[1])
             print(f'{total[0]}: {total[1]}')
+        plt.plot(x, y)
+        plt.savefig('ding.png')
 
 
 if __name__ == "__main__":
-- 
cgit