Repository: microsoft/retina
File: pkg/plugin/packetparser/packetparser_linux.go (Go)

Code before:
				uint16((bpfEvent.Flags&TCPFlagRST)>>2), // nolint:gomnd // 2 is the offset for RST.
				uint16((bpfEvent.Flags&TCPFlagPSH)>>3), // nolint:gomnd // 3 is the offset for PSH.
				uint16((bpfEvent.Flags&TCPFlagURG)>>5), // nolint:gomnd // 5 is the offset for URG.
			)

			// For packets originating from node, we use tsval as the tcpID.
			// Packets coming back has the tsval echoed in tsecr.
			if fl.GetTraceObservationPoint() == flow.TraceObservationPoint_TO_NETWORK {
				utils.AddTCPID(meta, uint64(tcpMetadata.Tsval))
			} else if fl.GetTraceObservationPoint() == flow.TraceObservationPoint_FROM_NETWORK {
				utils.AddTCPID(meta, uint64(tcpMetadata.Tsecr))
			}

			// Add metadata to the flow.
			utils.AddRetinaMetadata(fl, meta)

			// Write the event to the enricher.
			ev := &v1.Event{
				Event:     fl,
				Timestamp: fl.GetTime(),
			}
			if p.enricher != nil {
				p.enricher.Write(ev)
			}

			// Add conntrack metrics.
			if p.cfg.EnableConntrackMetrics {
				labels := []string{
					protoToString(bpfEvent.Proto),
					fl.GetTrafficDirection().String(),
				}
				// Basic metrics, node-level
				metrics.ConntrackPacketsForward.WithLabelValues(labels...).Set(float64(bpfEvent.ConntrackMetadata.PacketsForwardCount))
				metrics.ConntrackBytesForward.WithLabelValues(labels...).Set(float64(bpfEvent.ConntrackMetadata.BytesForwardCount))
				metrics.ConntrackPacketsReply.WithLabelValues(labels...).Set(float64(bpfEvent.ConntrackMetadata.PacketsReplyCount))
				metrics.ConntrackBytesReply.WithLabelValues(labels...).Set(float64(bpfEvent.ConntrackMetadata.BytesReplyCount))
			}

			// Write the event to the external channel.
			if p.externalChannel != nil {
				select {
				case p.externalChannel <- ev:
				default:
					// Channel is full, drop the event.
					// We shouldn't slow down the reader.
					metrics.LostEventsCounter.WithLabelValues(utils.ExternalChannel, name).Inc()
				}
			}
		}
	}
}

Review comment (suggestion):
this doesn't look right to me. If there are multiple connections of the same type and direction, when packetparser receives each packet event from each connection, would this just "reset" the packet and byte metrics for this node to whatever the counters of the connection the packet belongs to? Let's say that I have connection A and B, and packetparser is receiving packets from A for every other packet from B, the node metrics would flip flop

Code after:
				uint16((bpfEvent.Flags&TCPFlagACK)>>4), // nolint:gomnd // 4 is the offset for ACK.
				uint16((bpfEvent.Flags&TCPFlagFIN)>>0),
				uint16((bpfEvent.Flags&TCPFlagRST)>>2), // nolint:gomnd // 2 is the offset for RST.
				uint16((bpfEvent.Flags&TCPFlagPSH)>>3), // nolint:gomnd // 3 is the offset for PSH.
				uint16((bpfEvent.Flags&TCPFlagURG)>>5), // nolint:gomnd // 5 is the offset for URG.
			)

			// For packets originating from node, we use tsval as the tcpID.
			// Packets coming back has the tsval echoed in tsecr.
			if fl.GetTraceObservationPoint() == flow.TraceObservationPoint_TO_NETWORK {
				utils.AddTCPID(meta, uint64(tcpMetadata.Tsval))
			} else if fl.GetTraceObservationPoint() == flow.TraceObservationPoint_FROM_NETWORK {
				utils.AddTCPID(meta, uint64(tcpMetadata.Tsecr))
			}

			// Add metadata to the flow.
			utils.AddRetinaMetadata(fl, meta)

			// Write the event to the enricher.
			ev := &v1.Event{
				Event:     fl,
				Timestamp: fl.GetTime(),
			}
			if p.enricher != nil {
				p.enricher.Write(ev)
			}

			// Write the event to the external channel.
			if p.externalChannel != nil {
				select {
				case p.externalChannel <- ev:
				default:
					// Channel is full, drop the event.
					// We shouldn't slow down the reader.
					metrics.LostEventsCounter.WithLabelValues(utils.ExternalChannel, name).Inc()
				}
			}
		}
	}
}

func (p *packetParser) handleEvents(ctx context.Context) {
	for {
		select {
		case <-ctx.Done():
			p.l.Info("Context is done, stopping handleEvents")
			return
		default:
			p.readData()
		}
	}