diff options
| -rw-r--r-- | cleaner.go | 2 | ||||
| -rw-r--r-- | dbtestdata.sql | 14 | ||||
| -rw-r--r-- | sqlQueries.go | 14 |
3 files changed, 17 insertions, 13 deletions
@@ -80,7 +80,7 @@ func cleanData() (err error) { } func getTimespan(t time.Time) (span time.Time, err error) { - loc, err := time.LoadLocation("Local") + loc, err := time.LoadLocation(TIMEZONE) if err != nil { return } diff --git a/dbtestdata.sql b/dbtestdata.sql index c8553e2..7f877f1 100644 --- a/dbtestdata.sql +++ b/dbtestdata.sql @@ -1,11 +1,11 @@ ---New data that should not be processed in case there is more that should be aggregated together with it. +-- New data that should not be processed in case there is more that should be aggregated together with it. INSERT INTO raw_data (ip_src, ip_dst, time, port, packet_size) VALUES ("123.123.123.123", "12.12.12.12", NOW(), 80, 255); INSERT INTO raw_data (ip_src, ip_dst, time, port, packet_size) VALUES ("123.123.123.123", "12.12.12.12", NOW(), 80, 255); INSERT INTO raw_data (ip_src, ip_dst, time, port, packet_size) VALUES ("45.45.45.45", "12.12.12.12", NOW(), 80, 255); ---Old data that should be processed -INSERT INTO raw_data (ip_src, ip_dst, time, port, packet_size) VALUES ("63.63.63.63", "73.73.73.73", ('2008-12-01 12:00:00'), 80, 200); -INSERT INTO raw_data (ip_src, ip_dst, time, port, packet_size) VALUES ("63.63.63.63", "73.73.73.73", ('2008-12-01 12:00:00'), 80, 200); -INSERT INTO raw_data (ip_src, ip_dst, time, port, packet_size) VALUES ("8.8.8.8", "11.73.73.73", ('2008-12-01 12:00:00'), 80, 200); -INSERT INTO raw_data (ip_src, ip_dst, time, port, packet_size) VALUES ("63.63.63.63", "73.73.73.73", ('2008-12-01 12:00:00'), 80, 200); -INSERT INTO raw_data (ip_src, ip_dst, time, port, packet_size) VALUES ("100.63.63.63", "8.8.8.8", ('2008-12-01 12:00:00'), 80, 200); +-- Old data that should be processed +INSERT INTO raw_data (ip_src, ip_dst, time, port, packet_size) VALUES ("63.63.63.63", "73.73.73.73", '2008-12-01 12:00:00', 80, 200); +INSERT INTO raw_data (ip_src, ip_dst, time, port, packet_size) VALUES ("63.63.63.63", "73.73.73.73", '2008-12-01 12:00:00', 80, 200); +INSERT INTO raw_data (ip_src, ip_dst, time, port, packet_size) VALUES ("8.8.8.8", "11.73.73.73", '2008-12-01 12:00:00', 80, 200); +INSERT INTO raw_data (ip_src, ip_dst, time, port, packet_size) VALUES ("63.63.63.63", "73.73.73.73", '2008-12-01 12:00:00', 80, 200); +INSERT INTO raw_data (ip_src, ip_dst, time, port, packet_size) VALUES ("100.63.63.63", "8.8.8.8", '2008-12-01 12:00:00', 80, 200); diff --git a/sqlQueries.go b/sqlQueries.go index 086e780..af4f469 100644 --- a/sqlQueries.go +++ b/sqlQueries.go @@ -7,6 +7,10 @@ import ( "time" ) +const ( + TIMEZONE = "UTC" +) + //Retrieves all rawdata older than tim func fetchRawData(db *sql.DB, tim time.Time) (rDat []RawData, err error) { @@ -28,18 +32,18 @@ func fetchRawData(db *sql.DB, tim time.Time) (rDat []RawData, err error) { return } - prepUp, err := tx.Prepare("UPDATE raw_data SET process_time = ? where ip_src = ? AND ip_dst = ? AND time = ? AND port = ? and packet_size = ? LIMIT 1") + prepUp, err := tx.Prepare("UPDATE raw_data SET process_time = ? where ip_src = ? AND ip_dst = ? AND time = ? AND port = ? and packet_size = ? AND process_time IS NULL LIMIT 1") if err != nil { log.Println("Failed to prepare update") return } - loc, err := time.LoadLocation("Local") + loc, err := time.LoadLocation(TIMEZONE) for rows.Next() { var r RawData var tim []byte err = rows.Scan(&r.ipSrc, &r.ipDst, &tim, &r.port, &r.packetSize) - r.time, err = time.ParseInLocation("2006-02-01 15:04:05", string(tim), loc) + r.time, err = time.ParseInLocation("2006-01-02 15:04:05", string(tim), loc) if err != nil { log.Println("Failed to scan result of query") return @@ -95,14 +99,14 @@ func purgeRawData(tx *sql.Tx, rDat []RawData) (err error) { return } -func insertCleanData(tx *sql.Tx, ipbSrc, ipbDst, volume string, time time.Time, port, occurences int) error { +func insertCleanData(tx *sql.Tx, ipbSrc, ipbDst, volume string, tim time.Time, port, occurences int) error { prepStmt, err := tx.Prepare("INSERT INTO clean_data (ipb_src, ipb_dst, time, port, volume, occurences) VALUES ( ? , ? , ? , ? , ? , ?) ON DUPLICATE KEY UPDATE occurences = occurences + ?") if err != nil { log.Println("Failed to prepare statement") return err } - _, err = prepStmt.Exec(ipbSrc, ipbDst, time, port, volume, occurences, occurences) + _, err = prepStmt.Exec(ipbSrc, ipbDst, tim, port, volume, occurences, occurences) if err != nil { log.Println("Failed to execute statement") return err |