Big Refactor

README.md

@@ -23,6 +23,15 @@ MAXMIND_LICENSE_KEY=
 INFLUX_TOKEN=
 ```
 
+## NGINX
+
+The NGINX aggregator requires this `access_log` configuration:
+
+```nginx
+log_format config '"$time_local" "$remote_addr" "$request" "$status" "$body_bytes_sent" "$request_length" "$http_user_agent"';
+access_log /var/log/nginx/access.log config;
+```
+
 ## Dependencies
 
 Quick-Fedora-Mirror requires `zsh`

aggregator.go → aggregator/aggregator.go

@@ -1,4 +1,4 @@
-package main
+package aggregator
 
 import (
 	"time"
@@ -18,9 +18,14 @@ type Aggregator[T any] interface {
 	Send(writer api.WriteAPI)
 }
 
-// StartAggregator starts the aggregator with the given Aggregator implementation, channel of type T, influxdb QueryAPI and WriteAPI.
-// It returns the lastUpdated time and an error if any occurred during initialization.
-func StartAggregator[T any](aggregator Aggregator[T], c <-chan T) (lastUpdated time.Time, err error) {
+// StartAggregator is a function that starts an aggregator process to continuously
+// read data from a channel, aggregate it using the provided aggregator, and
+// send the aggregated data to a writer at regular intervals.
+//
+// # It takes an influxdb reader, writer, an aggregator over type T, and a channel of type T
+//
+// It returns the time of the last update (from influxdb), which can be used as a filter
+func StartAggregator[T any](reader api.QueryAPI, writer api.WriteAPI, aggregator Aggregator[T], c <-chan T) (lastUpdated time.Time, err error) {
 	lastUpdated, err = aggregator.Init(reader)
 	if err != nil {
 		return lastUpdated, err

aggregator_nginx.go → aggregator/aggregator_nginx.go

@@ -1,4 +1,4 @@
-package main
+package aggregator
 
 import (
 	"bufio"
@@ -17,6 +17,7 @@ import (
 	"time"
 
 	"github.com/COSI-Lab/Mirror/logging"
+	"github.com/COSI-Lab/geoip"
 	"github.com/IncSW/geoip2"
 	influxdb2 "github.com/influxdata/influxdb-client-go/v2"
 	"github.com/influxdata/influxdb-client-go/v2/api"
@@ -79,7 +80,6 @@ func (aggregator *NGINXProjectAggregator) Init(reader api.QueryAPI) (lastUpdated
 			result, err = reader.Query(context.Background(), request)
 
 			if err != nil {
-				logging.Warn("Failed to querying influxdb nginx statistics", err)
 				time.Sleep(time.Second)
 				continue
 			}
@@ -88,7 +88,7 @@ func (aggregator *NGINXProjectAggregator) Init(reader api.QueryAPI) (lastUpdated
 		}
 
 		if err != nil {
-			return lastUpdated, errors.New("error querying influxdb")
+			return lastUpdated, err
 		}
 
 		stats := make(ProjectStatistics)
@@ -207,7 +207,7 @@ func (aggregator *NGINXProjectAggregator) Send(writer api.WriteAPI) {
 	}
 }
 
-// NGINXLogEntry is a struct that represents a parsed nginx log entry
+// NGINXLogEntry represents a parsed nginx log entry
 type NGINXLogEntry struct {
 	IP        net.IP
 	City      *geoip2.CityResult
@@ -225,7 +225,7 @@ type NGINXLogEntry struct {
 var reQuotes = regexp.MustCompile(`"(.*?)"`)
 
 // TailNGINXLogFile tails a log file and sends the parsed log entries to the specified channels
-func TailNGINXLogFile(logFile string, lastUpdated time.Time, channels []chan<- NGINXLogEntry) {
+func TailNGINXLogFile(logFile string, lastUpdated time.Time, channels []chan<- NGINXLogEntry, geoipHandler *geoip.GeoIPHandler) {
 	start := time.Now()
 
 	f, err := os.Open(logFile)
@@ -261,7 +261,7 @@ func TailNGINXLogFile(logFile string, lastUpdated time.Time, channels []chan<- N
 
 	// Parse each line as we receive it
 	for line := range tail.Lines {
-		entry, err := parseNginxLine(line.Text)
+		entry, err := parseNginxLine(geoipHandler, line.Text)
 
 		if err == nil {
 			for ch := range channels {
@@ -280,7 +280,7 @@ func parseNginxDate(line string) (time.Time, error) {
 // It's critical the log file uses the correct format found at the top of this file
 // If the log file is not in the correct format or if some other part of the parsing fails
 // this function will return an error
-func parseNginxLine(line string) (entry NGINXLogEntry, err error) {
+func parseNginxLine(geoipHandler *geoip.GeoIPHandler, line string) (entry NGINXLogEntry, err error) {
 	// "$time_local" "$remote_addr" "$request" "$status" "$body_bytes_sent" "$request_length" "$http_user_agent";
 	quoteList := reQuotes.FindAllString(line, -1)
 
@@ -305,8 +305,8 @@ func parseNginxLine(line string) (entry NGINXLogEntry, err error) {
 		return entry, errors.New("failed to parse ip")
 	}
 
-	// Optional GeoIP lookup
-	if geoipHandler != nil {
+	// GeoIP lookup
+	if geoipHandler == nil {
 		city, err := geoipHandler.Lookup(entry.IP)
 		if err != nil {
 			entry.City = nil

aggregator_rsyncd.go → aggregator/aggregator_rsyncd.go

@@ -1,9 +1,8 @@
-package main
+package aggregator
 
 import (
 	"bufio"
 	"context"
-	"errors"
 	"fmt"
 	"io"
 	"os"
@@ -45,7 +44,6 @@ func (a *RSYNCDAggregator) Init(reader api.QueryAPI) (lastUpdated time.Time, err
 		result, err = reader.Query(context.Background(), request)
 
 		if err != nil {
-			logging.Warn("Failed to querying influxdb rsyncd statistics", err)
 			time.Sleep(time.Second)
 			continue
 		}
@@ -54,7 +52,7 @@ func (a *RSYNCDAggregator) Init(reader api.QueryAPI) (lastUpdated time.Time, err
 	}
 
 	if result == nil {
-		return time.Time{}, errors.New("Error querying influxdb for rsyncd stat")
+		return time.Time{}, err
 	}
 
 	for result.Next() {

aggregators.go

@@ -0,0 +1,61 @@
+package main
+
+import (
+	"net"
+	"time"
+
+	"github.com/COSI-Lab/Mirror/aggregator"
+	"github.com/COSI-Lab/Mirror/config"
+	"github.com/COSI-Lab/Mirror/logging"
+	"github.com/influxdata/influxdb-client-go/v2/api"
+)
+
+func StartNGINXAggregator(reader api.QueryAPI, writer api.WriteAPI, config *config.File) (chan<- aggregator.NGINXLogEntry, time.Time, error) {
+	nginxAg := aggregator.NewNGINXProjectAggregator()
+	nginxAg.AddMeasurement("nginx", func(re aggregator.NGINXLogEntry) bool {
+		return true
+	})
+
+	// Add subnet aggregators
+	for name, subnetStrings := range config.Subnets {
+		subnets := make([]*net.IPNet, 0)
+		for _, subnetString := range subnetStrings {
+			_, subnet, err := net.ParseCIDR(subnetString)
+			if err != nil {
+				logging.Warnf("Failed to parse subnet %q for %q", subnetString, name)
+				continue
+			}
+			subnets = append(subnets, subnet)
+		}
+
+		if len(subnets) == 0 {
+			logging.Warn("No valid subnets for", name)
+			continue
+		}
+
+		nginxAg.AddMeasurement(name, func(re aggregator.NGINXLogEntry) bool {
+			for _, subnet := range subnets {
+				if subnet.Contains(re.IP) {
+					return true
+				}
+			}
+			return false
+		})
+
+		logging.Infof("Added subnet aggregator for %q", name)
+	}
+
+	nginxMetrics := make(chan aggregator.NGINXLogEntry)
+	nginxLastUpdated, err := aggregator.StartAggregator[aggregator.NGINXLogEntry](reader, writer, nginxAg, nginxMetrics)
+
+	return nginxMetrics, nginxLastUpdated, err
+}
+
+func StartRSYNCAggregator(reader api.QueryAPI, writer api.WriteAPI) (chan<- aggregator.RSCYNDLogEntry, time.Time, error) {
+	rsyncAg := aggregator.NewRSYNCProjectAggregator()
+
+	rsyncMetrics := make(chan aggregator.RSCYNDLogEntry)
+	rsyncLastUpdated, err := aggregator.StartAggregator[aggregator.RSCYNDLogEntry](reader, writer, rsyncAg, rsyncMetrics)
+
+	return rsyncMetrics, rsyncLastUpdated, err
+}

config/configFile.go

@@ -8,7 +8,6 @@ import (
 	"os"
 	"sort"
 	"strings"
-	"text/template"
 
 	"github.com/xeipuuv/gojsonschema"
 )
@@ -101,46 +100,6 @@ func (config *File) GetProjects() []Project {
 	return projects
 }
 
-// CreateRSCYNDConfig writes a rsyncd.conf file to the given writer based on the config
-//
-// Consider passing a bufio.Write to this function
-func (config *File) CreateRSCYNDConfig(w io.Writer) error {
-	tmpl := `# This is a generated file. Do not edit manually.
-	uid = nobody
-	gid = nogroup
-	use chroot = yes
-	max connections = 0
-	pid file = /var/run/rsyncd.pid
-	motd file = /etc/rsyncd.motd
-	log file = /var/log/rsyncd.log
-	log format = %t %o %a %m %f %b
-	dont compress = *.gz *.tgz *.zip *.z *.Z *.rpm *.deb *.bz2 *.tbz2 *.xz *.txz *.rar
-	refuse options = checksum delete
-	{{ range . }}
-	[{{ .Short }}]
-		comment = {{ .Name }}
-		path = /storage/{{ .Short }}
-		exclude = lost+found/
-		read only = true
-		ignore nonreadable = yes{{ end }}
-	`
-
-	var filteredProjects []*Project
-	for _, project := range config.Projects {
-		if project.PublicRsync {
-			filteredProjects = append(filteredProjects, project)
-		}
-	}
-
-	t := template.Must(template.New("rsyncd.conf").Parse(tmpl))
-	err := t.Execute(w, filteredProjects)
-	if err != nil {
-		return err
-	}
-
-	return nil
-}
-
 // Validate checks the config file for a few properties
 //
 // - All projects have a unique long name, case insensitive

config/rsyncd.go

@@ -0,0 +1,44 @@
+package config
+
+import (
+	"io"
+	"text/template"
+)
+
+// CreateRSCYNDConfig writes a rsyncd.conf file to the given writer based on the Config struct
+func (config *File) CreateRSCYNDConfig(w io.Writer) error {
+	tmpl := `# This is a generated file. Do not edit manually.
+	uid = nobody
+	gid = nogroup
+	use chroot = yes
+	max connections = 0
+	pid file = /var/run/rsyncd.pid
+	motd file = /etc/rsyncd.motd
+	log file = /var/log/rsyncd.log
+	log format = %t %o %a %m %f %b
+	dont compress = *.gz *.tgz *.zip *.z *.Z *.rpm *.deb *.bz2 *.tbz2 *.xz *.txz *.rar
+	refuse options = checksum delete
+	{{ range . }}
+	[{{ .Short }}]
+		comment = {{ .Name }}
+		path = /storage/{{ .Short }}
+		exclude = lost+found/
+		read only = true
+		ignore nonreadable = yes{{ end }}
+	`
+
+	var filteredProjects []*Project
+	for _, project := range config.Projects {
+		if project.PublicRsync {
+			filteredProjects = append(filteredProjects, project)
+		}
+	}
+
+	t := template.Must(template.New("rsyncd.conf").Parse(tmpl))
+	err := t.Execute(w, filteredProjects)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}

config/tokens_test.go

@@ -7,7 +7,7 @@ import (
 	"github.com/COSI-Lab/Mirror/config"
 )
 
-// Test tokens.txt parsing
+// Test tokens.toml parsing
 func TestTokens(t *testing.T) {
 	example := `
 		[[tokens]]

influx.go

@@ -7,17 +7,12 @@ import (
 	"github.com/influxdata/influxdb-client-go/v2/api"
 )
 
-var writer api.WriteAPI
-var reader api.QueryAPI
-
 // SetupInfluxClients connects to influxdb and sets up the db clients
-func SetupInfluxClients(token string) {
+func SetupInfluxClients(token string) (reader api.QueryAPI, writer api.WriteAPI) {
 	// create new client with default option for server url authenticate by token
 	options := influxdb2.DefaultOptions()
 	options.SetTLSConfig(&tls.Config{InsecureSkipVerify: true})
 
 	client := influxdb2.NewClientWithOptions("https://mirror.clarkson.edu:8086", token, options)
-
-	writer = client.WriteAPI("COSI", "stats")
-	reader = client.QueryAPI("COSI")
+	return client.QueryAPI("COSI"), client.WriteAPI("COSI", "stats")
 }