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
|
package main
import (
"database/sql"
_ "github.com/go-sql-driver/mysql"
"time"
)
const (
DATABASE_USER = "root"
DATABASE_PASS = "pass"
DATABASE_CONNECTION = "" //e.g. "tcp(localhost:55555)
DATABASE_NAME = "netflow"
PROCESS_BATCH = 3
)
type RawData struct {
ipSrc string
ipDst string
time time.Time
port int
packetSize int
}
func derp() (err error) {
db, err := sql.Open("mysql", DATABASE_USER+":"+DATABASE_PASS+"@/"+DATABASE_NAME)
if err != nil {
return
}
defer db.Close()
rDat, err := fetchRawData(db)
if err != nil {
return
}
//Begin transaction
tx, err := db.Begin()
if err != nil {
return
}
//TODO insertCleanedData
//remove old data
err = purgeRawData(tx, rDat)
if err != nil {
tx.Rollback()
return
}
tx.Commit()
return
}
func insertCleanedData(tx *sql.DB, rDat []RawData) (err error) {
return
}
func fetchRawData(db *sql.DB) (rDat []RawData, err error) {
prepStmt, err := db.Prepare("SELECT * FROM raw_data LIMIT ? ")
if err != nil {
return
}
rows, err := prepStmt.Query(PROCESS_BATCH)
if err != nil {
return
}
for rows.Next() {
var r RawData
err = rows.Scan(&r.ipSrc, &r.ipDst, &r.time, &r.port, &r.packetSize)
if err != nil {
return
}
rDat = append(rDat, r)
}
return
}
func purgeRawData(tx *sql.Tx, rDat []RawData) (err error) {
for _, r := range rDat {
prepStmt, err := tx.Prepare("DELETE FROM raw_data WHERE ip_src = ? AND ip_dst = ? AND time = ? AND port = ? AND packet_size = ?")
if err != nil {
return
}
_, err = prepStmt.Exec(r.ipSrc, r.ipDst, r.time, r.port, r.packetSize)
if err != nil {
return
}
}
return
}
func insertCleanData(db *sql.DB, ipbSrc, ipbDst, volume string, time time.Time, port, occurences int) error {
prepStmt, err := db.Prepare("INSERT INTO clean_data (ipb_src, ipb_dst, time, port, volume, occurenaces) VALUES ( ? , ? , ? , ? , ? , ? , ?) ON DUPLICATE KEY UPDATE occurences = occurences + ?")
if err != nil {
return err
}
_, err = prepStmt.Exec(ipbSrc, ipbDst, time, port, volume, occurences, occurences)
if err != nil {
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
}
|
