diff options
Diffstat (limited to 'connection_handler.go')
| -rw-r--r-- | connection_handler.go | 49 |
1 files changed, 49 insertions, 0 deletions
diff --git a/connection_handler.go b/connection_handler.go index ffe4fac..4e92ccf 100644 --- a/connection_handler.go +++ b/connection_handler.go @@ -1,7 +1,25 @@ +/* + * This file is part of caronte (https://github.com/eciavatta/caronte). + * Copyright (c) 2020 Emiliano Ciavatta. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, version 3. + * + * This program is distributed in the hope that it will be useful, but + * WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see <http://www.gnu.org/licenses/>. + */ + package main import ( "encoding/binary" + "fmt" "github.com/flier/gohs/hyperscan" "github.com/google/gopacket" "github.com/google/gopacket/tcpassembly" @@ -225,6 +243,37 @@ func (ch *connectionHandlerImpl) Complete(handler *StreamHandler) { log.WithError(err).WithField("connection", connection).Error("failed to update all connections streams") } } + + ch.UpdateStatistics(connection) +} + +func (ch *connectionHandlerImpl) UpdateStatistics(connection Connection) { + rangeStart := connection.StartedAt.Unix() / 60 // group statistic records by minutes + duration := connection.ClosedAt.Sub(connection.StartedAt) + // if one of the two parts doesn't close connection, the duration is +infinity or -infinity + if duration.Hours() > 1 || duration.Hours() < -1 { + duration = 0 + } + servicePort := connection.DestinationPort + + updateDocument := UnorderedDocument{ + fmt.Sprintf("connections_per_service.%d", servicePort): 1, + fmt.Sprintf("client_bytes_per_service.%d", servicePort): connection.ClientBytes, + fmt.Sprintf("server_bytes_per_service.%d", servicePort): connection.ServerBytes, + fmt.Sprintf("total_bytes_per_service.%d", servicePort): connection.ClientBytes + connection.ServerBytes, + fmt.Sprintf("duration_per_service.%d", servicePort): duration.Milliseconds(), + } + + for _, ruleID := range connection.MatchedRules { + updateDocument[fmt.Sprintf("matched_rules.%s", ruleID.Hex())] = 1 + } + + var results interface{} + if _, err := ch.Storage().Update(Statistics).Upsert(&results). + Filter(OrderedDocument{{"_id", time.Unix(rangeStart*60, 0)}}). + OneComplex(UnorderedDocument{"$inc": updateDocument}); err != nil { + log.WithError(err).WithField("connection", connection).Error("failed to update connection statistics") + } } func (ch *connectionHandlerImpl) Storage() Storage { |