joewitt commented on code in PR #7830: URL: https://github.com/apache/nifi/pull/7830#discussion_r1343470578
########## nifi-nar-bundles/nifi-opentelemetry-bundle/nifi-opentelemetry-processors/src/main/java/org/apache/nifi/processors/opentelemetry/io/StandardRequestContentListener.java: ########## @@ -0,0 +1,269 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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. + */ +package org.apache.nifi.processors.opentelemetry.io; + +import com.google.protobuf.Message; +import io.opentelemetry.proto.collector.logs.v1.ExportLogsServiceRequest; +import io.opentelemetry.proto.collector.metrics.v1.ExportMetricsServiceRequest; +import io.opentelemetry.proto.collector.trace.v1.ExportTraceServiceRequest; +import io.opentelemetry.proto.common.v1.AnyValue; +import io.opentelemetry.proto.common.v1.KeyValue; +import io.opentelemetry.proto.logs.v1.ResourceLogs; +import io.opentelemetry.proto.metrics.v1.ResourceMetrics; +import io.opentelemetry.proto.resource.v1.Resource; +import io.opentelemetry.proto.trace.v1.ResourceSpans; +import org.apache.nifi.logging.ComponentLog; +import org.apache.nifi.processors.opentelemetry.encoding.JsonServiceRequestReader; +import org.apache.nifi.processors.opentelemetry.encoding.ProtobufServiceRequestReader; +import org.apache.nifi.processors.opentelemetry.encoding.ServiceRequestReader; +import org.apache.nifi.processors.opentelemetry.protocol.ServiceRequestDescription; +import org.apache.nifi.processors.opentelemetry.protocol.ServiceResponse; +import org.apache.nifi.processors.opentelemetry.protocol.ServiceResponseStatus; +import org.apache.nifi.processors.opentelemetry.protocol.TelemetryContentEncoding; +import org.apache.nifi.processors.opentelemetry.protocol.TelemetryContentType; +import org.apache.nifi.processors.opentelemetry.protocol.TelemetryRequestType; +import org.apache.nifi.stream.io.StreamUtils; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Objects; +import java.util.concurrent.BlockingQueue; +import java.util.zip.GZIPInputStream; + +/** + * Standard implementation of OTLP Request Content Listener supporting available Request and Content Types + */ +public class StandardRequestContentListener implements RequestContentListener { + private static final int ZERO_MESSAGES = 0; + + private static final byte COMPRESSED = 1; + + private static final String CLIENT_SOCKET_ADDRESS = "client.socket.address"; + + private static final String CLIENT_SOCKET_PORT = "client.socket.port"; + + private final ServiceRequestReader protobufReader = new ProtobufServiceRequestReader(); + + private final ServiceRequestReader jsonReader = new JsonServiceRequestReader(); + + private final ComponentLog log; + + private final BlockingQueue<Message> messages; + + public StandardRequestContentListener(final ComponentLog log, final BlockingQueue<Message> messages) { + this.log = Objects.requireNonNull(log, "Log required"); + this.messages = Objects.requireNonNull(messages, "Messages required"); + } + + /** + * Process Request Buffer supporting serialized Protobuf or JSON + * + * @param buffer Request Content Buffer to be processed + * @param serviceRequestDescription Service Request Description describes reader attributes + * @return Service Response with processing status + */ + @Override + public ServiceResponse onRequest(final ByteBuffer buffer, final ServiceRequestDescription serviceRequestDescription) { + Objects.requireNonNull(buffer, "Buffer required"); + Objects.requireNonNull(serviceRequestDescription, "Description required"); + + ServiceResponse serviceResponse; + + final InetSocketAddress remoteAddress = serviceRequestDescription.getRemoteAddress(); + final TelemetryContentType contentType = serviceRequestDescription.getContentType(); + if (TelemetryContentType.APPLICATION_GRPC == contentType) { + try { + final byte compression = buffer.get(); + final int messageSize = buffer.getInt(); + log.debug("Client Address [{}] Content-Type [{}] Message Size [{}] Compression [{}]", remoteAddress, contentType, messageSize, compression); + + final ByteBuffer messageBuffer; + if (COMPRESSED == compression) { + messageBuffer = getDecompressedBuffer(buffer); Review Comment: are we protected from this being potentially quite large? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@nifi.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
