CollectingServiceImpl implements CollectingService { @Override public Map.Entry<Optional<Long>, Optional<String>> collect(byte[] data, String remoteAddress, Optional<String> optionalHostName) throws NotEnoughBytesException, CollectingProcessException { final Optional<String>[] arrOptHostNames = new Optional[]{optionalHostName}; String deviceName = null; if (optionalHostName.isPresent()) { deviceName = optionalHostName.get(); } IpfixLogging ipfixLogging = new IpfixLogging(); IpfixMessage ipfixMessage = createIpfixMessage(data); ipfixLogging.setMessageHeader(ipfixMessage.getHeader()); LOGGER.debug(String.format("IPFIX message decoded successfully. Packet decoded: %s", ipfixMessage)); String messageExportTime = ipfixMessage.getHeader().getExportTime(); validateIpfixMessageExportTime(messageExportTime, remoteAddress, deviceName); long obsvDomain = ipfixMessage.getHeader().getObservationDomainId(); IpfixMessageNotification notificationMessage = new IpfixMessageNotification(); notificationMessage.setSourceIP(remoteAddress); notificationMessage.setObservationDomain(obsvDomain); notificationMessage.setTimestamp(messageExportTime); Map<Integer, AbstractTemplateRecord> templateByIdMap = buildTemplateToIdMap(ipfixMessage); if (arrOptHostNames[0].isPresent() && !templateByIdMap.isEmpty()) { m_cachingService.cacheTemplateRecord(obsvDomain, arrOptHostNames[0].get(), templateByIdMap); } final boolean[] needToCache = {!arrOptHostNames[0].isPresent()}; LOGGER.debug(String.format("Hostname %s and templateByIdMap %s", arrOptHostNames[0], templateByIdMap)); TemplateProvider templateProvider = templateId -> { if (arrOptHostNames[0].isPresent()) { try { AbstractTemplateRecord templateCache = m_cachingService.getTemplateRecord(obsvDomain, arrOptHostNames[0].get(), templateId); if (Objects.isNull(templateCache)) { needToCache[0] = true; LOGGER.debug(String.format("Cannot get template cache %d for device %s.", templateId, arrOptHostNames[0].get())); } return templateCache; } catch (Exception e) { LOGGER.debug(String.format("Error while retrieving template cache %d for device %s.", templateId, arrOptHostNames[0].get())); } } LOGGER.debug(String.format("Hostname isn't present. Need to cache the templates %s", templateByIdMap)); return templateByIdMap.get(templateId); }; List<IpfixDataSet> cacheDataSetFromPreviousMessage = new ArrayList<>(); if (arrOptHostNames[0].isPresent()) { cacheDataSetFromPreviousMessage = m_cachingService.getAndInvalidateIpfixDataSet(obsvDomain, arrOptHostNames[0].get()); } List<IpfixDataSet> rawDataSets = new LinkedList<>(cacheDataSetFromPreviousMessage); if (ipfixMessage.getDataSets() != null) { List<AbstractTemplateRecord> templateRecords = extractTemplateRecordFromMessage(ipfixMessage); if (!templateRecords.isEmpty()) { ipfixLogging.addTemplateSet(ipfixMessage.getTemplateSets()); } List<AbstractTemplateRecord> optionTemplateRecords = extractOptionTemplateRecordFromMessage(ipfixMessage); if (!optionTemplateRecords.isEmpty()) { ipfixLogging.addOptionTemplateSet(ipfixMessage.getOptionTemplateSets()); } LOGGER.debug(String.format("Decoded more packages from buffer: %s", rawDataSets)); rawDataSets.addAll(ipfixMessage.getDataSets()); } for (IpfixDataSet set : rawDataSets) { Set<IpfixDecodedData> decodeDataSet; try { decodeDataSet = m_decodingDataRecordService.decodeDataSet(arrOptHostNames[0].orElse(""), obsvDomain, set, ipfixMessage, templateProvider); } catch (DecodingException e) { throw new CollectingProcessException("Error while decoding data set.", e); } boolean isOptionTemplateSet = false; if (decodeDataSet != null) { isOptionTemplateSet = true; } IpfixSetHeader setHeader = set.getHeader(); IpfixLoggingDataSet ipfixLoggingDataSet = new IpfixLoggingDataSet(setHeader); ipfixLoggingDataSet.setDataRecords(set.getRecords(), isOptionTemplateSet); AbstractTemplateRecord abstractTemplateRecord = templateProvider.get(setHeader.getId()); if (abstractTemplateRecord != null) { ipfixLoggingDataSet.setTemplateRecords(abstractTemplateRecord.getFieldSpecifiers()); } ipfixLogging.addToSetMessages(ipfixLoggingDataSet); if (decodeDataSet != null && !decodeDataSet.isEmpty()) { int templateID = set.getHeader().getId(); LOGGER.debug(String.format("Data set decoded successfully (Observation domain: %s, Set id: %s)", obsvDomain, templateID)); LOGGER.info("Decoded ipfix message for " + deviceName + " is: " + ipfixLogging.convertToLog()); if (isAuthenticateDataSet(decodeDataSet)) { String hostName = ""; for (IpfixDecodedData decodedData : decodeDataSet) { if (decodedData.getCounterName().equalsIgnoreCase(IpfixConstants.HOSTNAME_IE_KEY)) { hostName = decodedData.getDecodedValue(); } } if (arrOptHostNames[0].isPresent() && !arrOptHostNames[0].get().equals(hostName)) { m_cachingService.updateHostName(obsvDomain, arrOptHostNames[0].get(), hostName); } if (hostName.isEmpty()) { throw new CollectingProcessException("Missing hostname"); } arrOptHostNames[0] = Optional.of(hostName); notificationMessage.setHostName(hostName); } else { if (arrOptHostNames[0].isPresent() && !arrOptHostNames[0].get().isEmpty()) { handleDataSetForTemplateSet(arrOptHostNames[0].get(), decodeDataSet, templateID, notificationMessage); } else { LOGGER.error(String.format("Couldn't find hostname so data set %s will be ignored", set.toString())); break; } } if (arrOptHostNames[0].isPresent()) { deviceName = arrOptHostNames[0].get(); } } else { LOGGER.debug(String.format("Buffering data set (Observation domain: %s, Set id: %s)", obsvDomain, set.getHeader().getId())); if (arrOptHostNames[0].isPresent()) { m_cachingService.cacheIpfixDataSet(obsvDomain, arrOptHostNames[0].get(), set); } else { LOGGER.info(String.format("Couldn't find hostname so data set %s will be ignored", set.toString())); } } } if (arrOptHostNames[0].isPresent() && needToCache[0]) { LOGGER.debug(String.format("Need to cache the templates %s for device %s", templateByIdMap, arrOptHostNames[0].get())); m_cachingService.cacheTemplateRecord(obsvDomain, arrOptHostNames[0].get(), templateByIdMap); } return new AbstractMap.SimpleEntry(Optional.of(obsvDomain), arrOptHostNames[0]); } CollectingServiceImpl(DecodingDataRecordService decodingDataRecordService, IpfixCachingService cachingService,
                                 DataHandlerService dataHandlerService, DeviceCacheService deviceCacheService); @Override Map.Entry<Optional<Long>, Optional<String>> collect(byte[] data, String remoteAddress, Optional<String> optionalHostName); @Override void processAuthenFailure(); @Override void processDecodeFailure(); @Override void processMessageCount();  }
@Test public void testCollectTemplateSet() throws Exception { byte[] data = IpfixUtilities.hexStringToByteArray("000A01DB5DB0B80D00000000000010EF00020020011400038107FFFF" + "00000E918109000400000E91810A000400000E91011401AB0444534C3100000001000000020544534C3130000000010000000" + "10544534C313100000001000000010544534C313200000001000000010544534C313300000001000000010544534C31340000" + "0001000000020544534C313500000001000000010544534C313600000001000000010444534C3200000001000000010844534" + "C322E50544D00000001000000010444534C3300000001000000010844534C332E50544D00000001000000010444534C340000" + "0001000000010444534C3500000001000000010444534C3600000001000000010444534C3700000001000000010444534C380" + "0000001000000010444534C39000000010000000109534650312E45544831000000010000000109534650322E455448310000" + "00010000000106696E62616E6400000001000000010C6E66616365312D75736572310000000100000001076F757462616E640" + "0000001000000010773696E62616E64000000010000000108736F757462616E6400000001000000010C7566616365312D7573" + "65723100000001000000010C7566616365312D757365723300000001000000010258310000000100000001"); IpfixMessage ipfixMessage = new IpfixMessage(data); ipfixMessage.getHeader().setExportTime(Instant.now().toString()); AbstractTemplateRecord templateRecord = new IpfixTemplateRecord(new byte[16]); m_cachingService = mock(IpfixCachingServiceImpl.class); when(m_cachingService.getTemplateRecord(4335L, "hostname", 600)).thenReturn(templateRecord); String hostName = "hostname"; String remoteAddress = "192.168.1.1"; String username = "adminuser"; String hashedPass = "$6$129267dbd52b96f3$BcacdzmrCHVqpQcEklp1ZjsBkW2n78xsTfl.nVKOFa.G1QXjC.MXtamcnAKCWRAMMg/rd/GCOSTq20T72CRGk."; Set<IpfixDecodedData> decodedDataSet = new HashSet<>(); decodedDataSet.add(new IpfixDecodedData("8636.3729", "String", username)); decodedDataSet.add(new IpfixDecodedData("8637.3729", "String", hashedPass)); decodedDataSet.add(new IpfixDecodedData("5002.3729", "String", hostName)); CollectingServiceImpl collectingService = new CollectingServiceImpl(m_decodingDataRecordService, m_cachingService, m_dataHandlerService, m_deviceFamilyCacheService) { @Override protected IpfixMessage createIpfixMessage(byte[] data) { return ipfixMessage; } }; String family = "sample-DPU-model-tls-1.0"; Optional<String> optionalHostName = Optional.of(hostName); when(m_deviceFamilyCacheService.getDeviceFamily(hostName)).thenReturn(family); when(m_decodingDataRecordService.decodeDataSet(eq(hostName), eq(4335L), any(), any(), any())).thenReturn(decodedDataSet); List<IpfixDataHandler> dataHandlers = new ArrayList<>(); IpfixDataHandler dataHandler = mock(IpfixDataHandler.class); dataHandlers.add(dataHandler); when(m_dataHandlerService.getDataHandlers()).thenReturn(dataHandlers); collectingService.collect(m_data, remoteAddress, optionalHostName); ArgumentMatcher<String> messageMatcher = new ArgumentMatcher<String>() { @Override public boolean matches(Object o) { String msgAsStr = (String) o; Gson gson = new Gson(); IpfixMessageNotification message = gson.fromJson(msgAsStr, IpfixMessageNotification.class); assertNotNull(message); assertNotNull(message.getTimestamp()); assertEquals("hostname", message.getHostName()); assertEquals("192.168.1.1", message.getSourceIP()); assertEquals("sample-DPU-model-tls-1.0", message.getDeviceAdapter()); assertEquals(276, message.getTemplateID()); assertEquals(4335L, message.getObservationDomain()); return true; } }; verify(dataHandler).handleIpfixData(argThat(messageMatcher)); }