|
| 1 | +// Copyright 2025 Antrea Authors |
| 2 | +// |
| 3 | +// Licensed under the Apache License, Version 2.0 (the "License"); |
| 4 | +// you may not use this file except in compliance with the License. |
| 5 | +// You may obtain a copy of the License at |
| 6 | +// |
| 7 | +// http://www.apache.org/licenses/LICENSE-2.0 |
| 8 | +// |
| 9 | +// Unless required by applicable law or agreed to in writing, software |
| 10 | +// distributed under the License is distributed on an "AS IS" BASIS, |
| 11 | +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 12 | +// See the License for the specific language governing permissions and |
| 13 | +// limitations under the License. |
| 14 | + |
| 15 | +package flowaggregator |
| 16 | + |
| 17 | +import ( |
| 18 | + "fmt" |
| 19 | + "net" |
| 20 | + |
| 21 | + "github.com/vmware/go-ipfix/pkg/entities" |
| 22 | + "k8s.io/klog/v2" |
| 23 | +) |
| 24 | + |
| 25 | +// preprocessor is in charge of processing messages received from the IPFIX collector, prior to |
| 26 | +// handling records over to the aggregation process. At the moment, its only task is to ensure that |
| 27 | +// all records have the expected fields. If a record has extra fields, they will be discarded. If |
| 28 | +// some fields are missing, they will be "appended" to the record with a "zero" value. For example, |
| 29 | +// we will use 0 for integral types, "" for strings, 0.0.0.0 for IPv4 address, etc. Note that we are |
| 30 | +// able to keep the implementation simple by assuming that a record either has missing fields or |
| 31 | +// extra fields (not a combination of both), and that such fields are always at the tail of the |
| 32 | +// field list. This assumption is based on implementation knowledge of the FlowExporter and the |
| 33 | +// FlowAggregator. |
| 34 | +type preprocessor struct { |
| 35 | + inCh <-chan *entities.Message |
| 36 | + outCh chan<- entities.Record |
| 37 | + |
| 38 | + expectedElementsV4 int |
| 39 | + expectedElementsV6 int |
| 40 | + |
| 41 | + defaultElementsWithValueV4 []entities.InfoElementWithValue |
| 42 | + defaultElementsWithValueV6 []entities.InfoElementWithValue |
| 43 | +} |
| 44 | + |
| 45 | +func makeDefaultElementWithValue(ie *entities.InfoElement) (entities.InfoElementWithValue, error) { |
| 46 | + switch ie.DataType { |
| 47 | + case entities.OctetArray: |
| 48 | + var val []byte |
| 49 | + if ie.Len < entities.VariableLength { |
| 50 | + val = make([]byte, ie.Len) |
| 51 | + } |
| 52 | + return entities.NewOctetArrayInfoElement(ie, val), nil |
| 53 | + case entities.Unsigned8: |
| 54 | + return entities.NewUnsigned8InfoElement(ie, 0), nil |
| 55 | + case entities.Unsigned16: |
| 56 | + return entities.NewUnsigned16InfoElement(ie, 0), nil |
| 57 | + case entities.Unsigned32: |
| 58 | + return entities.NewUnsigned32InfoElement(ie, 0), nil |
| 59 | + case entities.Unsigned64: |
| 60 | + return entities.NewUnsigned64InfoElement(ie, 0), nil |
| 61 | + case entities.Signed8: |
| 62 | + return entities.NewSigned8InfoElement(ie, 0), nil |
| 63 | + case entities.Signed16: |
| 64 | + return entities.NewSigned16InfoElement(ie, 0), nil |
| 65 | + case entities.Signed32: |
| 66 | + return entities.NewSigned32InfoElement(ie, 0), nil |
| 67 | + case entities.Signed64: |
| 68 | + return entities.NewSigned64InfoElement(ie, 0), nil |
| 69 | + case entities.Float32: |
| 70 | + return entities.NewFloat32InfoElement(ie, 0), nil |
| 71 | + case entities.Float64: |
| 72 | + return entities.NewFloat64InfoElement(ie, 0), nil |
| 73 | + case entities.Boolean: |
| 74 | + return entities.NewBoolInfoElement(ie, false), nil |
| 75 | + case entities.DateTimeSeconds: |
| 76 | + return entities.NewDateTimeSecondsInfoElement(ie, 0), nil |
| 77 | + case entities.DateTimeMilliseconds: |
| 78 | + return entities.NewDateTimeMillisecondsInfoElement(ie, 0), nil |
| 79 | + case entities.MacAddress: |
| 80 | + return entities.NewMacAddressInfoElement(ie, make([]byte, 6)), nil |
| 81 | + case entities.Ipv4Address: |
| 82 | + return entities.NewIPAddressInfoElement(ie, net.IPv4zero), nil |
| 83 | + case entities.Ipv6Address: |
| 84 | + return entities.NewIPAddressInfoElement(ie, net.IPv6zero), nil |
| 85 | + case entities.String: |
| 86 | + return entities.NewStringInfoElement(ie, ""), nil |
| 87 | + default: |
| 88 | + return nil, fmt.Errorf("unexpected Information Element data type: %d", ie.DataType) |
| 89 | + } |
| 90 | +} |
| 91 | + |
| 92 | +func makeDefaultElementsWithValue(infoElements []*entities.InfoElement) ([]entities.InfoElementWithValue, error) { |
| 93 | + elementsWithValue := make([]entities.InfoElementWithValue, len(infoElements)) |
| 94 | + for idx := range infoElements { |
| 95 | + var err error |
| 96 | + if elementsWithValue[idx], err = makeDefaultElementWithValue(infoElements[idx]); err != nil { |
| 97 | + return nil, err |
| 98 | + } |
| 99 | + } |
| 100 | + return elementsWithValue, nil |
| 101 | +} |
| 102 | + |
| 103 | +func newPreprocessor(infoElementsV4, infoElementsV6 []*entities.InfoElement, inCh <-chan *entities.Message, outCh chan<- entities.Record) (*preprocessor, error) { |
| 104 | + defaultElementsWithValueV4, err := makeDefaultElementsWithValue(infoElementsV4) |
| 105 | + if err != nil { |
| 106 | + return nil, fmt.Errorf("error when generating default values for IPv4 Information Elements expected from exporter: %w", err) |
| 107 | + } |
| 108 | + defaultElementsWithValueV6, err := makeDefaultElementsWithValue(infoElementsV6) |
| 109 | + if err != nil { |
| 110 | + return nil, fmt.Errorf("error when generating default values for IPv6 Information Elements expected from exporter: %w", err) |
| 111 | + } |
| 112 | + return &preprocessor{ |
| 113 | + inCh: inCh, |
| 114 | + outCh: outCh, |
| 115 | + expectedElementsV4: len(infoElementsV4), |
| 116 | + expectedElementsV6: len(infoElementsV6), |
| 117 | + defaultElementsWithValueV4: defaultElementsWithValueV4, |
| 118 | + defaultElementsWithValueV6: defaultElementsWithValueV6, |
| 119 | + }, nil |
| 120 | +} |
| 121 | + |
| 122 | +func (p *preprocessor) Run(stopCh <-chan struct{}) { |
| 123 | + for { |
| 124 | + select { |
| 125 | + case <-stopCh: |
| 126 | + return |
| 127 | + case msg, ok := <-p.inCh: |
| 128 | + if !ok { |
| 129 | + return |
| 130 | + } |
| 131 | + p.processMsg(msg) |
| 132 | + } |
| 133 | + } |
| 134 | +} |
| 135 | + |
| 136 | +func isRecordIPv4(record entities.Record) bool { |
| 137 | + _, _, exist := record.GetInfoElementWithValue("sourceIPv4Address") |
| 138 | + return exist |
| 139 | +} |
| 140 | + |
| 141 | +func (p *preprocessor) processMsg(msg *entities.Message) { |
| 142 | + set := msg.GetSet() |
| 143 | + if set.GetSetType() != entities.Data { |
| 144 | + return |
| 145 | + } |
| 146 | + records := set.GetRecords() |
| 147 | + for _, record := range records { |
| 148 | + elementList := record.GetOrderedElementList() |
| 149 | + numElements := len(elementList) |
| 150 | + isIPv4 := isRecordIPv4(record) |
| 151 | + expectedElements := p.expectedElementsV4 |
| 152 | + if !isIPv4 { |
| 153 | + expectedElements = p.expectedElementsV6 |
| 154 | + } |
| 155 | + if numElements == expectedElements { |
| 156 | + p.outCh <- record |
| 157 | + } else if numElements > expectedElements { |
| 158 | + if klog.V(5).Enabled() { |
| 159 | + klog.InfoS("Record received from exporter includes unexpected elements, truncating", "expectedElements", expectedElements, "receivedElements", numElements) |
| 160 | + } |
| 161 | + // Creating a new Record seems like the best option here. By using |
| 162 | + // NewDataRecordFromElements, we should minimize the number of allocations |
| 163 | + // required. |
| 164 | + p.outCh <- entities.NewDataRecordFromElements(0, elementList[:expectedElements], true) |
| 165 | + } else { |
| 166 | + if klog.V(5).Enabled() { |
| 167 | + klog.InfoS("Record received from exporter is missing information elements, adding fields with zero values", "expectedElements", expectedElements, "receivedElements", numElements) |
| 168 | + } |
| 169 | + if isIPv4 { |
| 170 | + elementList = append(elementList, p.defaultElementsWithValueV4[numElements:]...) |
| 171 | + } else { |
| 172 | + elementList = append(elementList, p.defaultElementsWithValueV6[numElements:]...) |
| 173 | + } |
| 174 | + p.outCh <- entities.NewDataRecordFromElements(0, elementList, true) |
| 175 | + } |
| 176 | + } |
| 177 | + |
| 178 | +} |
0 commit comments