loki submit over otlp

[bilsch-nice]

We are in the process of setting up loki tempo and mimir and feeding data in via otel-collector. The data ingests via the otlp interfaces ( http or grpc ).

It would be nice to see the loki generator support submission via otlp in addition to the native loki connection. Not sure if this is in scope for the project or not

[gpcmol]

I want this feature too. Since I don't have go knowledge, I wrote my own converter to convert the log streams into OTLP. The idea here is to make use of the OpenTelemetry sdk.
It's just 1 file Spring / Kotlin application.

Change the base url in the k6 test to:

const BASE_URL = "http://logconverter.k6.svc.cluster.local:8081";

DemoApplication.kt

package com.example.demo

import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
import com.fasterxml.jackson.module.kotlin.readValue
import io.opentelemetry.api.common.AttributeKey
import io.opentelemetry.api.logs.LogRecordBuilder
import io.opentelemetry.exporter.otlp.http.logs.OtlpHttpLogRecordExporter
import io.opentelemetry.sdk.OpenTelemetrySdk
import io.opentelemetry.sdk.logs.SdkLoggerProvider
import io.opentelemetry.sdk.logs.export.BatchLogRecordProcessor
import org.springframework.beans.factory.annotation.Value
import org.springframework.boot.autoconfigure.SpringBootApplication
import org.springframework.boot.runApplication
import org.springframework.http.ResponseEntity
import org.springframework.web.bind.annotation.PostMapping
import org.springframework.web.bind.annotation.RequestBody
import org.springframework.web.bind.annotation.RequestMapping
import org.springframework.web.bind.annotation.RestController
import java.time.Instant

@SpringBootApplication
class DemoApplication

fun main(args: Array<String>) {
    runApplication<DemoApplication>(*args)
}

data class LokiPayload(
    val streams: List<LokiStream>
)

data class LokiStream(
    val stream: Map<String, Any>,
    val values: List<List<String>>
)

@RestController
@RequestMapping("/loki/api/v1/push")
class LogConverterController(@Value("\${collectorEndpoint}") private val collectorEndpoint: String) {
    private val objectMapper: ObjectMapper = jacksonObjectMapper()

    val sdk = OpenTelemetrySdk.builder()
        .setLoggerProvider(
            SdkLoggerProvider.builder()
                .addLogRecordProcessor(
                    BatchLogRecordProcessor.builder(
                        OtlpHttpLogRecordExporter.builder().setEndpoint(collectorEndpoint).build()
                    ).build()
                ).build()
        ).build()

    @PostMapping
    fun convertToOtlp(@RequestBody lokiPayload: String): ResponseEntity<Void> {
        println(lokiPayload)

        val lokiData: LokiPayload = objectMapper.readValue(lokiPayload)

        lokiData.streams.forEach { stream ->
            stream.values.forEach { value ->
                val timestamp = Instant.ofEpochSecond(0, value[0].toLong())
                val message = value[1]

                val logRecord: LogRecordBuilder = sdk.sdkLoggerProvider.get("").logRecordBuilder()
                stream.stream.forEach { (key, value) ->
                    when (value) {
                        is String -> {
                            val attributeKey = AttributeKey.stringKey(key)
                            logRecord.setAttribute(attributeKey, value)
                        }

                        is Number -> {
                            val attributeKey = AttributeKey.doubleKey(key)
                            logRecord.setAttribute(attributeKey, value.toDouble())
                        }

                        is Boolean -> {
                            val attributeKey = AttributeKey.booleanKey(key)
                            logRecord.setAttribute(attributeKey, value)
                        }

                        else -> {
                            val attributeKey = AttributeKey.stringKey(key)
                            logRecord.setAttribute(attributeKey, value.toString())
                        }
                    }
                }

                logRecord
                    .setBody(message)
                    .setTimestamp(timestamp)

                logRecord.emit()
            }
        }
        return ResponseEntity.noContent().build()
    }

}

${opentelemetry.version} = 1.45.0
pom.xml

		<dependency>
			<groupId>io.opentelemetry</groupId>
			<artifactId>opentelemetry-exporter-otlp</artifactId>
			<version>${opentelemetry.version}</version>
		</dependency>
		<dependency>
			<groupId>io.opentelemetry</groupId>
			<artifactId>opentelemetry-exporter-logging</artifactId>
			<version>${opentelemetry.version}</version>
		</dependency>
		<dependency>
			<groupId>io.opentelemetry</groupId>
			<artifactId>opentelemetry-sdk</artifactId>
			<version>${opentelemetry.version}</version>
		</dependency>

application.properties

spring.application.name=demo
server.port=8081
collectorEndpoint=${COLLECTOR_ENDPOINT:http://opentelemetry-collector.telemetry.svc.cluster.local:4318/v1/logs}
spring.threads.virtual.enabled=true