1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
|
package main
import (
"database/sql"
_ "github.com/go-sql-driver/mysql"
"log"
"time"
)
//Retrieves all rawdata older than tim
func fetchRawData(db *sql.DB, tim time.Time) (rDat []RawData, err error) {
prepSel, err := db.Prepare("SELECT ip_src,ip_dst,time,port,packet_size FROM raw_data WHERE time < tim")
if err != nil {
log.Println("Failed to prepare select")
return
}
rows, err := prepSel.Query(numRows)
if err != nil {
log.Println("Failed to query prepared selection")
return
}
tx, err := db.Begin()
if err != nil {
log.Println("Failed to initialize transaction")
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")
if err != nil {
log.Println("Failed to prepare update")
return
}
loc, err := time.LoadLocation("Local")
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)
if err != nil {
log.Println("Failed to scan result of query")
return
}
prepUp.Exec(time.Now(), r.ipSrc, r.ipDst, r.time, r.port, r.packetSize)
if err != nil {
log.Println("Failed to query prepared update")
tx.Rollback()
return
}
rDat = append(rDat, r)
}
tx.Commit()
return
}
//Removes the process_time from every entry that started being proccesed before tim
func reprocess(db *sq.DB, tim time.Time) (err error) {
stmt, err := db.Prepare("UPDATE raw_data SET process_time = NULL WHERE process_time < ?")
if err != nil {
return
}
_, err = stmt.Execute(tim)
return
}
//Removes all entries in the database that started being processed before tim
func purgeProcessed(db *sql.DB, tim time.Time) (err error) {
stmt, err := db.Prepare("DELETE FROM raw_data WHERE process_time < ? ")
if err != nil {
return
}
_, err = stmt.Execute(tim)
return
}
func purgeRawData(tx *sql.Tx, rDat []RawData) (err error) {
prepStmt, err := tx.Prepare("DELETE FROM raw_data WHERE ip_src = ? AND ip_dst = ? AND time = ? AND port = ? AND packet_size = ? AND process_time IS NOT NULL LIMIT 1")
if err != nil {
return
}
for _, r := range rDat {
_, err = prepStmt.Exec(r.ipSrc, r.ipDst, r.time, r.port, r.packetSize)
if err != nil {
return
}
}
return
}
func insertCleanData(tx *sql.Tx, ipbSrc, ipbDst, volume string, time 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)
if err != nil {
log.Println("Failed to execute statement")
return err
}
return nil
}
func insertASNIP(db *sql.DB, asn int, ipBlock string) error {
prepCheck, err := db.Prepare("SELECT asn FROM asnip WHERE ip_block = ?")
if err != nil {
return err
}
defer prepCheck.Close()
rows, err := prepCheck.Exec(ipBlock)
if err != nil {
return err
}
if rows != nil {
return nil
}
prepIns, err := db.Prepare("INSERT INTO asnip VALUES ( ? , ? )")
if err != nil {
return err
}
defer prepIns.Close()
_, err = prepIns.Exec(asn, ipBlock)
if err != nil {
return err
}
return nil
}
func removeASNIP(db *sql.DB, asn int, ipBlock string) error {
prepStmt, err := db.Prepare("DELETE FROM anip WHERE asn = ? AND ip_block = ?")
if err != nil {
return err
}
defer prepStmt.Close()
_, err = prepStmt.Exec(asn, ipBlock)
if err != nil {
return err
}
return nil
}
|
