From 23438166832778b36f3aac7b3cc0b81ab6e8da1e Mon Sep 17 00:00:00 2001 From: fkrzywka Date: Tue, 5 Jun 2018 12:55:06 +0200 Subject: Extract domain module WireFrame extracted to domain module Domain package name in collector-core changed to avoid collision Merged Protobuf module with new domain module Closes ONAP-253. Change-Id: Ie7ac0bb877eae018a7520b4f3ceebe9fd51dbf65 Signed-off-by: fkrzywka Issue-ID: DCAEGEN2-601 --- hv-collector-domain/pom.xml | 128 +++++++++++++++++++++ .../onap/dcae/collectors/veshv/domain/WireFrame.kt | 93 +++++++++++++++ .../src/main/proto/HVRanMeasFields-v5.proto | 54 +++++++++ .../src/main/proto/VesEvent-v5.proto | 86 ++++++++++++++ .../dcae/collectors/veshv/domain/WireFrameTest.kt | 55 +++++++++ hv-collector-domain/src/test/resources/logback.xml | 54 +++++++++ 6 files changed, 470 insertions(+) create mode 100644 hv-collector-domain/pom.xml create mode 100644 hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/WireFrame.kt create mode 100644 hv-collector-domain/src/main/proto/HVRanMeasFields-v5.proto create mode 100644 hv-collector-domain/src/main/proto/VesEvent-v5.proto create mode 100644 hv-collector-domain/src/test/kotlin/org/onap/dcae/collectors/veshv/domain/WireFrameTest.kt create mode 100644 hv-collector-domain/src/test/resources/logback.xml (limited to 'hv-collector-domain') diff --git a/hv-collector-domain/pom.xml b/hv-collector-domain/pom.xml new file mode 100644 index 00000000..de9004a8 --- /dev/null +++ b/hv-collector-domain/pom.xml @@ -0,0 +1,128 @@ + + + + 4.0.0 + + + + The Apache Software License, Version 2.0 + http://www.apache.org/licenses/LICENSE-2.0.txt + + + + + org.onap.dcaegen2.collectors.veshv + ves-hv-collector + 1.0.0-SNAPSHOT + .. + + + hv-collector-domain + VES HighVolume Collector :: Domain + + + false + + + + + + kotlin-maven-plugin + org.jetbrains.kotlin + + + org.apache.maven.plugins + maven-compiler-plugin + + + maven-surefire-plugin + org.apache.maven.plugins + + + org.codehaus.mojo + build-helper-maven-plugin + + + com.github.os72 + protoc-jar-maven-plugin + + + generate-sources + + run + + + com.google.protobuf:protoc:${protobuf.version} + + ${project.basedir}/src/main/proto + + + + java + none + ${protobuf-generated-files.directory} + + + + + + + + + + + + org.jetbrains.kotlin + kotlin-stdlib-jdk8 + + + io.projectreactor.ipc + reactor-netty + + + com.google.protobuf + protobuf-java + + + + org.assertj + assertj-core + + + org.jetbrains.kotlin + kotlin-test + ${kotlin.version} + test + + + org.jetbrains.spek + spek-api + + + org.jetbrains.spek + spek-junit-platform-engine + + + + + diff --git a/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/WireFrame.kt b/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/WireFrame.kt new file mode 100644 index 00000000..906441b9 --- /dev/null +++ b/hv-collector-domain/src/main/kotlin/org/onap/dcae/collectors/veshv/domain/WireFrame.kt @@ -0,0 +1,93 @@ +/* + * ============LICENSE_START======================================================= + * dcaegen2-collectors-veshv + * ================================================================================ + * Copyright (C) 2018 NOKIA + * ================================================================================ + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * ============LICENSE_END========================================================= + */ +package org.onap.dcae.collectors.veshv.domain + +import io.netty.buffer.ByteBuf +import io.netty.buffer.ByteBufAllocator + +/** + * Wire frame structure is presented bellow. All fields are in network byte order (big-endian). + * + * ``` + * ┌─────┬───────────────────────┬───────────────────────┬───────────────────────┐ + * │octet│ 0 │ 1 │ 2 │ + * ├─────┼──┬──┬──┬──┬──┬──┬──┬──┼──┬──┬──┬──┬──┬──┬──┬──┼──┬──┬──┬──┬──┬──┬──┬──┤ + * │ bit │ 0│ │ │ │ │ │ │ │ 8│ │ │ │ │ │ │ │16│ │ │ │ │ │ │ │ ... + * ├─────┼──┴──┴──┴──┴──┴──┴──┴──┼──┴──┴──┴──┴──┴──┴──┴──┼──┴──┴──┴──┴──┴──┴──┴──┤ + * │field│ 0xFF │ major version │ minor version │ + * └─────┴───────────────────────┴───────────────────────┴───────────────────────┘ + * ┌─────┬───────────────────────┬───────────────────────┬───────────────────────┬───────────────────────┐ + * │octet│ 3 │ 4 │ 5 │ 6 │ + * ├─────┼──┬──┬──┬──┬──┬──┬──┬──┼──┬──┬──┬──┬──┬──┬──┬──┼──┬──┬──┬──┬──┬──┬──┬──┼──┬──┬──┬──┬──┬──┬──┬──┤ + * ... │ bit │24│ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ ... + * ├─────┼──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┴──┤ + * │field│ payload size │ + * └─────┴───────────────────────────────────────────────────────────────────────────────────────────────┘ + * ┌─────┬─────────────────────── + * │octet│ 7 ... + * ├─────┼──┬──┬──┬──┬──┬──┬──┬── + * ... │ bit │56│ │ │ │ │ │ │... + * ├─────┼──┴──┴──┴──┴──┴──┴──┴── + * │field│ protobuf payload + * └─────┴─────────────────────── + * ``` + * + * @author Piotr Jaszczyk + * @since May 2018 + */ +data class WireFrame(val payload: ByteBuf, + val mark: Short, + val majorVersion: Short, + val minorVersion: Short, + val payloadSize: Int) { + + fun isValid(): Boolean = + mark == FF_BYTE + && majorVersion == SUPPORTED_MAJOR_VERSION + && payload.readableBytes() == payloadSize + + fun encode(allocator: ByteBufAllocator): ByteBuf { + val bb = allocator.buffer(HEADER_SIZE + payload.readableBytes()) + + bb.writeByte(mark.toInt()) + bb.writeByte(majorVersion.toInt()) + bb.writeByte(minorVersion.toInt()) + bb.writeInt(payloadSize) + bb.writeBytes(payload) + + return bb + } + + companion object { + fun decode(byteBuf: ByteBuf): WireFrame { + val mark = byteBuf.readUnsignedByte() + val majorVersion = byteBuf.readUnsignedByte() + val minorVersion = byteBuf.readUnsignedByte() + val payloadSize = byteBuf.readInt() + val payload = byteBuf.retainedSlice() + + return WireFrame(payload, mark, majorVersion, minorVersion, payloadSize) + } + + private const val HEADER_SIZE = 3 + java.lang.Integer.BYTES + private const val FF_BYTE: Short = 0xFF + private const val SUPPORTED_MAJOR_VERSION: Short = 1 + } +} diff --git a/hv-collector-domain/src/main/proto/HVRanMeasFields-v5.proto b/hv-collector-domain/src/main/proto/HVRanMeasFields-v5.proto new file mode 100644 index 00000000..5121f0eb --- /dev/null +++ b/hv-collector-domain/src/main/proto/HVRanMeasFields-v5.proto @@ -0,0 +1,54 @@ +/* + * ============LICENSE_START======================================================= + * dcaegen2-collectors-veshv + * ================================================================================ + * Copyright (C) 2018 NOKIA + * ================================================================================ + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * ============LICENSE_END========================================================= + */ +syntax = "proto3"; +package org.onap.ves; + +// Definition for RTPM + +message HVRanMeasFields { + message HVRanMeasPayload { + message PMObject { + message HVRanMeas { + uint32 measurement_id = 1; + repeated uint32 counter_subid = 2; + repeated sint64 counter_value = 3; + repeated uint32 missing_counter_subid = 4; + bool suspectFlagIncomplete = 5; // (some is data missing due to internal error) + bool suspectFlagOutOfSync = 6; // (source time not aligned) + } + + string uri = 1; // monitored object URI + repeated HVRanMeas hvRanMeas = 2; // performance counters grouped by measurement types + } + repeated PMObject pmObject = 1; + } + + message AdditionalField { + string name = 1; + string value = 2; + } + + string hvRanMeasFieldsVersion = 1; // version of HVRanMeasFields message + uint32 period_ms = 2; // period configured for reporting the data in milliseconds + string timezone = 3; // timezone of Network Function sending the data + string pmDictionaryVsn = 4; // vendor name + schema version E.g. NOKIA_LN7.0, uniquely identify the relevant PM dictionary + HVRanMeasPayload hvRanMeasPayload = 5; // objects being monitored + repeated AdditionalField additionalFields = 6; // array of name-value pairs if needed +} \ No newline at end of file diff --git a/hv-collector-domain/src/main/proto/VesEvent-v5.proto b/hv-collector-domain/src/main/proto/VesEvent-v5.proto new file mode 100644 index 00000000..022cce4e --- /dev/null +++ b/hv-collector-domain/src/main/proto/VesEvent-v5.proto @@ -0,0 +1,86 @@ +/* + * ============LICENSE_START======================================================= + * dcaegen2-collectors-veshv + * ================================================================================ + * Copyright (C) 2018 NOKIA + * ================================================================================ + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * ============LICENSE_END========================================================= + */ +syntax = "proto3"; +package org.onap.ves; + +message VesEvent { + + // VES CommonEventHeader adapted to GPB (Google Protocol Buffers) + // Source: https://git.opnfv.org/ves/tree/tests/docs/ves_data_model.json + // 2017-05-13 Align with VES 5.0 schema. + // blob: ca948ff67e8a2de4e2a47cffc4d4d2893170ab76 + + message CommonEventHeader { + string version = 1; // required, "version of the event header" + enum Domain { + FAULT = 0; + HEARTBEAT = 1; + MEASUREMENTS_FOR_VF_SCALING = 2; + MOBILE_FLOW = 3; + SIP_SIGNALING = 4; + STATE_CHANGE = 5; + SYSLOG = 6; + THRESHOLD_CROSSING_ALERT = 7; + VOICE_QUALITY = 8; + OTHER = 9; + HVRANMEAS = 10; + } + Domain domain = 2; // required, "the eventing domain associated with the event" [map to string] + + uint32 sequence = 3; // required, "ordering of events communicated by an event source instance or 0 if not needed" + + enum Priority { + HIGH = 0; + MEDIUM = 1; + NORMAL = 2; + LOW = 3; + } + Priority priority = 4; // required, "processing priority" + + string eventId = 5; // required, "event key that is unique to the event source" + string eventName = 6; // required, "unique event name" + string eventType = 7; // "for example - applicationVnf, guestOS, hostOS, platform" + + uint64 lastEpochMicrosec = 8; // required, "the latest unix time aka epoch time associated with the event from any component--as microseconds elapsed since 1 Jan 1970 not including leap seconds" + uint64 startEpochMicrosec = 9; // required, "the earliest unix time aka epoch time associated with the event from any component--as microseconds elapsed since 1 Jan 1970 not including leap seconds" + + string nfNamingCode = 10; // "4 character network function type, aligned with vnf naming standards" + string nfcNamingCode = 11; // "3 character network function component type, aligned with vfc naming standards" + + string reportingEntityId = 12; // "UUID identifying the entity reporting the event, for example an OAM VM; must be populated by the ATT enrichment process" + bytes reportingEntityName = 13; // required, "name of the entity reporting the event, for example, an EMS name; may be the same as sourceName" + bytes sourceId = 14; // "UUID identifying the entity experiencing the event issue; must be populated by the ATT enrichment process" + string sourceName = 15; // required, "name of the entity experiencing the event issue" + + reserved "InternalHeaderFields"; // "enrichment fields for internal VES Event Listener service use only, not supplied by event sources" + reserved 100; + } + + CommonEventHeader commonEventHeader = 1; + + oneof eventFields // required, payload, each high-volume domain has its specific GPB schema + { + bytes hvRanMeasFields = 2; // if domain==HVRANMEAS, GPB schema: HVRanMeasFields.proto + } +} + +message VesEventList { + repeated VesEvent vesEvent = 1; +} diff --git a/hv-collector-domain/src/test/kotlin/org/onap/dcae/collectors/veshv/domain/WireFrameTest.kt b/hv-collector-domain/src/test/kotlin/org/onap/dcae/collectors/veshv/domain/WireFrameTest.kt new file mode 100644 index 00000000..5a923c4e --- /dev/null +++ b/hv-collector-domain/src/test/kotlin/org/onap/dcae/collectors/veshv/domain/WireFrameTest.kt @@ -0,0 +1,55 @@ +package org.onap.dcae.collectors.veshv.domain + +import io.netty.buffer.ByteBufAllocator +import io.netty.buffer.Unpooled +import org.assertj.core.api.Assertions.assertThat +import org.jetbrains.spek.api.Spek +import org.jetbrains.spek.api.dsl.describe +import org.jetbrains.spek.api.dsl.it + +/** + * @author Piotr Jaszczyk + * @since June 2018 + */ +object WireFrameTest : Spek({ + describe("Wire Frame codec") { + describe("encode-decode methods' compatibility") { + val payloadContent = "test" + val payload = Unpooled.wrappedBuffer(payloadContent.toByteArray(Charsets.US_ASCII)) + val frame = WireFrame(payload = payload, + majorVersion = 1, + minorVersion = 2, + mark = 0xFF, + payloadSize = payload.readableBytes()) + + val encoded = frame.encode(ByteBufAllocator.DEFAULT) + val decoded = WireFrame.decode(encoded) + + it("should decode major version") { + assertThat(decoded.majorVersion).isEqualTo(frame.majorVersion) + } + + it("should decode minor version") { + assertThat(decoded.minorVersion).isEqualTo(frame.minorVersion) + } + + it("should decode mark") { + assertThat(decoded.mark).isEqualTo(frame.mark) + } + + it("should decode payload size") { + assertThat(decoded.payloadSize).isEqualTo(frame.payloadSize) + } + + it("should decode payload") { + assertThat(decoded.payload.toString(Charsets.US_ASCII)) + .isEqualTo(payloadContent) + } + + it("should retain decoded payload") { + encoded.release() + assertThat(decoded.payload.refCnt()).isEqualTo(1) + } + } + } +}) \ No newline at end of file diff --git a/hv-collector-domain/src/test/resources/logback.xml b/hv-collector-domain/src/test/resources/logback.xml new file mode 100644 index 00000000..0bf2cb02 --- /dev/null +++ b/hv-collector-domain/src/test/resources/logback.xml @@ -0,0 +1,54 @@ + + + + + + + + + + %d{yyyy-MM-dd'T'HH:mm:ss.SSSXXX,UTC} %highlight(%-5level) [%-40.40logger{10}] - %msg%n + + + + + + + ${FILE_LOG_PATTERN} + + ${LOG_FILE} + + ${LOG_FILE}.%d{yyyy-MM-dd}.log + 50MB + 30 + 10GB + + + + + + + + + + \ No newline at end of file -- cgit 1.2.3-korg