mirror of
https://github.com/mii443/prometheus-android-exporter.git
synced 2025-08-22 15:15:35 +00:00
code formatting
This commit is contained in:
Martin Ptáček
@ -14,8 +14,8 @@ private const val TAG: String = "CONFIGURATION"
|
||||
//TODO register within prometheus foundation
|
||||
private const val defaultPrometheusServerPort: Int = 10101
|
||||
private const val defaultRemoteWriteScrapeInterval: Int = 30 // seconds
|
||||
private const val defaultRemoteWriteMaxSamplesPerExport : Int = 60 // seconds
|
||||
private const val defaultRemoteWriteExportInterval : Int = 120 // seconds
|
||||
private const val defaultRemoteWriteMaxSamplesPerExport: Int = 60 // seconds
|
||||
private const val defaultRemoteWriteExportInterval: Int = 120 // seconds
|
||||
|
||||
// serialization classes for parsing YAML configuration file
|
||||
@Serializable
|
||||
@ -27,15 +27,19 @@ data class PromConfigFile(
|
||||
fun toPromConfiguration(): PromConfiguration {
|
||||
return PromConfiguration(
|
||||
pushproxProxyUrl = this.pushprox?.proxy_url ?: "",
|
||||
remoteWriteScrapeInterval = (this.remote_write?.scrape_interval ?: defaultRemoteWriteScrapeInterval).toString(),
|
||||
remoteWriteScrapeInterval = (this.remote_write?.scrape_interval
|
||||
?: defaultRemoteWriteScrapeInterval).toString(),
|
||||
pushproxEnabled = this.pushprox?.enabled ?: false,
|
||||
pushproxFqdn = this.pushprox?.fqdn ?: "",
|
||||
remoteWriteEnabled = this.remote_write?.enabled ?: false,
|
||||
remoteWriteEndpoint = this.remote_write?.remote_write_endpoint ?: "",
|
||||
prometheusServerEnabled = this.prometheus_server?.enabled ?: true,
|
||||
prometheusServerPort = (this.prometheus_server?.port ?: defaultPrometheusServerPort).toString(),
|
||||
remoteWriteMaxSamplesPerExport = (this.remote_write?.max_samples_per_export ?: defaultRemoteWriteMaxSamplesPerExport).toString(),
|
||||
remoteWriteExportInterval = (this.remote_write?.export_interval ?: defaultRemoteWriteExportInterval).toString(),
|
||||
prometheusServerPort = (this.prometheus_server?.port
|
||||
?: defaultPrometheusServerPort).toString(),
|
||||
remoteWriteMaxSamplesPerExport = (this.remote_write?.max_samples_per_export
|
||||
?: defaultRemoteWriteMaxSamplesPerExport).toString(),
|
||||
remoteWriteExportInterval = (this.remote_write?.export_interval
|
||||
?: defaultRemoteWriteExportInterval).toString(),
|
||||
)
|
||||
}
|
||||
}
|
||||
@ -74,8 +78,8 @@ data class PromConfiguration(
|
||||
val remoteWriteEnabled: Boolean = false,
|
||||
val remoteWriteScrapeInterval: String = defaultRemoteWriteScrapeInterval.toString(),
|
||||
val remoteWriteEndpoint: String = "",
|
||||
val remoteWriteExportInterval : String = defaultRemoteWriteExportInterval.toString(),
|
||||
val remoteWriteMaxSamplesPerExport : String = defaultRemoteWriteMaxSamplesPerExport.toString(),
|
||||
val remoteWriteExportInterval: String = defaultRemoteWriteExportInterval.toString(),
|
||||
val remoteWriteMaxSamplesPerExport: String = defaultRemoteWriteMaxSamplesPerExport.toString(),
|
||||
) {
|
||||
|
||||
fun toStructuredText(): String {
|
||||
|
||||
@ -27,11 +27,8 @@ import androidx.compose.material.TopAppBar
|
||||
import androidx.compose.material.icons.Icons
|
||||
import androidx.compose.material.icons.outlined.Settings
|
||||
import androidx.compose.runtime.Composable
|
||||
import androidx.compose.runtime.MutableState
|
||||
import androidx.compose.runtime.collectAsState
|
||||
import androidx.compose.runtime.getValue
|
||||
import androidx.compose.runtime.mutableStateOf
|
||||
import androidx.compose.runtime.remember
|
||||
import androidx.compose.ui.Alignment
|
||||
import androidx.compose.ui.Modifier
|
||||
import androidx.compose.ui.graphics.Color
|
||||
@ -301,7 +298,7 @@ private fun RemoteWritePage(
|
||||
horizontalAlignment = Alignment.CenterHorizontally,
|
||||
) {
|
||||
Text("Remote write configuration:")
|
||||
|
||||
|
||||
Spacer(modifier = Modifier.padding(bottom = 12.dp))
|
||||
|
||||
TextField(
|
||||
|
||||
@ -65,7 +65,7 @@ class PromViewModel : ViewModel() {
|
||||
private lateinit var getContext: () -> Context
|
||||
|
||||
private var workerLiveData: LiveData<List<WorkInfo>>? = null
|
||||
private val workerLiveDataObserver : Observer<List<WorkInfo>> = Observer {
|
||||
private val workerLiveDataObserver: Observer<List<WorkInfo>> = Observer {
|
||||
if (it.isEmpty()) {
|
||||
updateExporterStateWith(ExporterState.NotRunning)
|
||||
} else {
|
||||
@ -83,6 +83,7 @@ class PromViewModel : ViewModel() {
|
||||
companion object {
|
||||
private const val PROM_UNIQUE_WORK: String = "prom_unique_job"
|
||||
}
|
||||
|
||||
private fun loadConfigurationFile() {
|
||||
Log.v(TAG, "Checking for configuration file")
|
||||
|
||||
@ -139,18 +140,18 @@ class PromViewModel : ViewModel() {
|
||||
}
|
||||
}
|
||||
|
||||
override fun onCleared(){
|
||||
override fun onCleared() {
|
||||
super.onCleared()
|
||||
workerLiveData?.removeObserver(workerLiveDataObserver)
|
||||
}
|
||||
|
||||
private fun updateExporterStateWith(exporterState: ExporterState){
|
||||
private fun updateExporterStateWith(exporterState: ExporterState) {
|
||||
_uiState.update {
|
||||
it.copy(exporterState = exporterState)
|
||||
}
|
||||
}
|
||||
|
||||
private fun startMonitoringWorker(){
|
||||
private fun startMonitoringWorker() {
|
||||
val workManagerInstance = WorkManager.getInstance(getContext())
|
||||
workerLiveData = workManagerInstance.getWorkInfosLiveData(
|
||||
WorkQuery.fromUniqueWorkNames(
|
||||
@ -185,15 +186,15 @@ class PromViewModel : ViewModel() {
|
||||
return false
|
||||
}
|
||||
|
||||
private fun somePushProxVariableUnset(config : PromConfiguration) : Boolean {
|
||||
private fun somePushProxVariableUnset(config: PromConfiguration): Boolean {
|
||||
return config.pushproxFqdn.isBlank() || config.pushproxProxyUrl.isBlank()
|
||||
}
|
||||
|
||||
private fun somePrometheusServerVariableUnset(config : PromConfiguration) : Boolean {
|
||||
private fun somePrometheusServerVariableUnset(config: PromConfiguration): Boolean {
|
||||
return config.prometheusServerPort.isBlank()
|
||||
}
|
||||
|
||||
private fun someRemoteWriteVariableUnset(config : PromConfiguration) : Boolean {
|
||||
private fun someRemoteWriteVariableUnset(config: PromConfiguration): Boolean {
|
||||
return config.remoteWriteEndpoint.isBlank()
|
||||
|| config.remoteWriteScrapeInterval.isBlank()
|
||||
|| config.remoteWriteExportInterval.isBlank()
|
||||
@ -209,22 +210,22 @@ class PromViewModel : ViewModel() {
|
||||
}
|
||||
|
||||
// check for empty configuration
|
||||
if(config.pushproxEnabled && somePushProxVariableUnset(config)){
|
||||
if (config.pushproxEnabled && somePushProxVariableUnset(config)) {
|
||||
return displayConfigValidationDialog("Please set all PushProx configuration settings!")
|
||||
}
|
||||
if(config.prometheusServerEnabled && somePrometheusServerVariableUnset(config)){
|
||||
if (config.prometheusServerEnabled && somePrometheusServerVariableUnset(config)) {
|
||||
return displayConfigValidationDialog("Set all Prometheus Server config settings!")
|
||||
}
|
||||
if(config.remoteWriteEnabled && someRemoteWriteVariableUnset(config)){
|
||||
if (config.remoteWriteEnabled && someRemoteWriteVariableUnset(config)) {
|
||||
return displayConfigValidationDialog("Set all Remote Write configuration settings!")
|
||||
}
|
||||
|
||||
// validate settings for remote write
|
||||
if(config.remoteWriteEnabled){
|
||||
if (config.remoteWriteEnabled) {
|
||||
// check scrape interval boundaries
|
||||
val minScrapeInterval = 1
|
||||
val maxScrapeInterval = 3600 / 4
|
||||
val scrapeInterval : Int = config.remoteWriteScrapeInterval.toIntOrNull()
|
||||
val scrapeInterval: Int = config.remoteWriteScrapeInterval.toIntOrNull()
|
||||
?: return displayConfigValidationDialog("Scrape interval must be a number!")
|
||||
|
||||
if (scrapeInterval > maxScrapeInterval || scrapeInterval < minScrapeInterval) {
|
||||
@ -236,9 +237,9 @@ class PromViewModel : ViewModel() {
|
||||
?: return displayConfigValidationDialog("Max Samples Per Export must be a number!")
|
||||
|
||||
// check export interval
|
||||
val exportInterval : Int = config.remoteWriteExportInterval.toIntOrNull()
|
||||
val exportInterval: Int = config.remoteWriteExportInterval.toIntOrNull()
|
||||
?: return displayConfigValidationDialog("Export interval must be a number!")
|
||||
if (scrapeInterval > exportInterval){
|
||||
if (scrapeInterval > exportInterval) {
|
||||
return displayConfigValidationDialog(
|
||||
"Scrape interval must be smaller than Export interval!"
|
||||
)
|
||||
@ -247,11 +248,11 @@ class PromViewModel : ViewModel() {
|
||||
}
|
||||
|
||||
// validate settings for prometheus server
|
||||
if(config.prometheusServerEnabled){
|
||||
if (config.prometheusServerEnabled) {
|
||||
// check port boundaries
|
||||
val minPort = 1024
|
||||
val maxPort = 65535
|
||||
val prometheusServerPort : Int = config.prometheusServerPort.toIntOrNull()
|
||||
val prometheusServerPort: Int = config.prometheusServerPort.toIntOrNull()
|
||||
?: return displayConfigValidationDialog("Prometheus Server Port must be a number!")
|
||||
if (prometheusServerPort < minPort || prometheusServerPort > maxPort) {
|
||||
return displayConfigValidationDialog("Prometheus exporter port out of bounds!")
|
||||
|
||||
@ -11,8 +11,10 @@ class ExponentialBackoff {
|
||||
private const val initialDelay = 3.0 // seconds
|
||||
private const val maxDelay = 40.0 // seconds
|
||||
|
||||
suspend fun runWithBackoff(function: suspend () -> Unit,
|
||||
onException: () -> Unit, infinite : Boolean = true) {
|
||||
suspend fun runWithBackoff(
|
||||
function: suspend () -> Unit,
|
||||
onException: () -> Unit, infinite: Boolean = true
|
||||
) {
|
||||
|
||||
var successfull: Boolean = false
|
||||
|
||||
@ -41,7 +43,7 @@ class ExponentialBackoff {
|
||||
currentDelay = min(currentDelay, maxDelay)
|
||||
|
||||
// finite vs infinite exponential backoff
|
||||
if(currentDelay == maxDelay && !infinite){
|
||||
if (currentDelay == maxDelay && !infinite) {
|
||||
break
|
||||
}
|
||||
|
||||
|
||||
@ -27,7 +27,8 @@ class PromWorker(
|
||||
|
||||
private val collectorRegistry = CollectorRegistry()
|
||||
private val metricsEngine: MetricsEngine = MetricsEngine(context)
|
||||
private val androidCustomExporter : AndroidCustomExporter = AndroidCustomExporter(metricsEngine).register(collectorRegistry)
|
||||
private val androidCustomExporter: AndroidCustomExporter =
|
||||
AndroidCustomExporter(metricsEngine).register(collectorRegistry)
|
||||
private lateinit var pushProxClient: PushProxClient
|
||||
private var remoteWriteSender: RemoteWriteSender? = null
|
||||
|
||||
@ -42,16 +43,16 @@ class PromWorker(
|
||||
return writer.toString()
|
||||
}
|
||||
|
||||
private suspend fun countSuccessfulScrape(){
|
||||
private suspend fun countSuccessfulScrape() {
|
||||
remoteWriteSender?.countSuccessfulScrape()
|
||||
}
|
||||
|
||||
@OptIn(DelicateCoroutinesApi::class)
|
||||
private suspend fun startServicesInOneThread(config: PromConfiguration){
|
||||
private suspend fun startServicesInOneThread(config: PromConfiguration) {
|
||||
val threadContext = newSingleThreadContext("PromWorkerThreadContext")
|
||||
|
||||
coroutineScope {
|
||||
withContext(threadContext){
|
||||
withContext(threadContext) {
|
||||
|
||||
if (config.remoteWriteEnabled) {
|
||||
remoteWriteSender = RemoteWriteSender(
|
||||
|
||||
@ -28,34 +28,35 @@ private const val TAG: String = "REMOTE_WRITE_SENDER"
|
||||
//
|
||||
// Only timestamps of succesfull scrapes are stored
|
||||
private class LastTimeRingBuffer(private val scrapeIntervalMs: Int) {
|
||||
private val buffer : Array<Long> = Array(hysteresisThreshold) { 0 }
|
||||
private var firstIndex : Int = 0
|
||||
companion object{
|
||||
private const val hysteresisThreshold : Int = 3
|
||||
private val buffer: Array<Long> = Array(hysteresisThreshold) { 0 }
|
||||
private var firstIndex: Int = 0
|
||||
|
||||
companion object {
|
||||
private const val hysteresisThreshold: Int = 3
|
||||
}
|
||||
|
||||
fun setLastTime(timestamp : Long) {
|
||||
fun setLastTime(timestamp: Long) {
|
||||
firstIndex = firstIndex++ % hysteresisThreshold
|
||||
buffer[firstIndex] = timestamp
|
||||
}
|
||||
|
||||
private fun getTimeByIndex(index : Int) : Long {
|
||||
if(index > hysteresisThreshold - 1){
|
||||
private fun getTimeByIndex(index: Int): Long {
|
||||
if (index > hysteresisThreshold - 1) {
|
||||
throw IllegalArgumentException("index cannot be bigger than hysteresisThreshold")
|
||||
}
|
||||
|
||||
val bufferIndex : Int = firstIndex + index % hysteresisThreshold
|
||||
val bufferIndex: Int = firstIndex + index % hysteresisThreshold
|
||||
return buffer[bufferIndex]
|
||||
}
|
||||
|
||||
fun checkScrapeDidNotHappenInTime() : Boolean {
|
||||
fun checkScrapeDidNotHappenInTime(): Boolean {
|
||||
return getTimeByIndex(0) < System.currentTimeMillis() - 3 * scrapeIntervalMs
|
||||
}
|
||||
|
||||
fun checkScrapeDidNotHappenHysteresis() : Boolean {
|
||||
val scrapeOccurredAfterThis : Long = System.currentTimeMillis() - 5 * scrapeIntervalMs
|
||||
fun checkScrapeDidNotHappenHysteresis(): Boolean {
|
||||
val scrapeOccurredAfterThis: Long = System.currentTimeMillis() - 5 * scrapeIntervalMs
|
||||
for (i in 0 until hysteresisThreshold) {
|
||||
if (getTimeByIndex(i) < scrapeOccurredAfterThis){
|
||||
if (getTimeByIndex(i) < scrapeOccurredAfterThis) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
@ -69,33 +70,33 @@ data class RemoteWriteConfiguration(
|
||||
val remoteWriteEndpoint: String,
|
||||
val collectorRegistry: CollectorRegistry,
|
||||
val maxSamplesPerExport: Int,
|
||||
val exportInterval : Int,
|
||||
val getContext : () -> Context,
|
||||
val exportInterval: Int,
|
||||
val getContext: () -> Context,
|
||||
)
|
||||
|
||||
class RemoteWriteSender(private val config: RemoteWriteConfiguration) {
|
||||
private val lastTimeRingBuffer = LastTimeRingBuffer(config.scrapeInterval * 1000)
|
||||
private val storage : RemoteWriteSenderStorage = RemoteWriteSenderSimpleMemoryStorage()
|
||||
private var scrapesAreBeingSent : Boolean = false
|
||||
private lateinit var client : HttpClient
|
||||
private var lastTimeRemoteWriteSent : Long = 0
|
||||
private var remoteWriteOn : Boolean = false
|
||||
private val storage: RemoteWriteSenderStorage = RemoteWriteSenderSimpleMemoryStorage()
|
||||
private var scrapesAreBeingSent: Boolean = false
|
||||
private lateinit var client: HttpClient
|
||||
private var lastTimeRemoteWriteSent: Long = 0
|
||||
private var remoteWriteOn: Boolean = false
|
||||
|
||||
private suspend fun performScrapeAndSaveIt(channel : Channel<Unit>) {
|
||||
private suspend fun performScrapeAndSaveIt(channel: Channel<Unit>) {
|
||||
val scrapedMetrics = config.collectorRegistry.metricFamilySamples()
|
||||
storage.writeScrapedSample(scrapedMetrics)
|
||||
channel.send(Unit)
|
||||
}
|
||||
|
||||
private suspend fun scraper(channel : Channel<Unit>){
|
||||
private suspend fun scraper(channel: Channel<Unit>) {
|
||||
val checkDelay = 1000L
|
||||
while (true){
|
||||
if (lastTimeRingBuffer.checkScrapeDidNotHappenInTime()){
|
||||
while (true) {
|
||||
if (lastTimeRingBuffer.checkScrapeDidNotHappenInTime()) {
|
||||
remoteWriteOn = true
|
||||
performScrapeAndSaveIt(channel)
|
||||
delay(config.scrapeInterval * 1000L)
|
||||
|
||||
while(lastTimeRingBuffer.checkScrapeDidNotHappenHysteresis()){
|
||||
while (lastTimeRingBuffer.checkScrapeDidNotHappenHysteresis()) {
|
||||
delay(config.scrapeInterval * 1000L)
|
||||
performScrapeAndSaveIt(channel)
|
||||
}
|
||||
@ -104,44 +105,44 @@ class RemoteWriteSender(private val config: RemoteWriteConfiguration) {
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun sendAll(){
|
||||
private suspend fun sendAll() {
|
||||
scrapesAreBeingSent = true
|
||||
while (!storage.isEmpty()){
|
||||
while (!storage.isEmpty()) {
|
||||
val body = storage.getScrapedSamplesCompressedProtobuf(config.maxSamplesPerExport)
|
||||
ExponentialBackoff.runWithBackoff( {sendRequestToRemoteWrite(body)}, {}, false)
|
||||
ExponentialBackoff.runWithBackoff({ sendRequestToRemoteWrite(body) }, {}, false)
|
||||
}
|
||||
lastTimeRemoteWriteSent = System.currentTimeMillis()
|
||||
}
|
||||
|
||||
private fun deviceHasInternet() : Boolean {
|
||||
private fun deviceHasInternet(): Boolean {
|
||||
val connectivityManager = config.getContext()
|
||||
.getSystemService(Context.CONNECTIVITY_SERVICE) as ConnectivityManager?
|
||||
|
||||
if (connectivityManager != null){
|
||||
if (connectivityManager != null) {
|
||||
val network = connectivityManager.getActiveNetworkCompat()
|
||||
val cap = connectivityManager.getNetworkCapabilities(network)
|
||||
if (cap != null && cap.hasCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET)){
|
||||
if (cap != null && cap.hasCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET)) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
private fun timeHasPassed() : Boolean {
|
||||
private fun timeHasPassed(): Boolean {
|
||||
return lastTimeRemoteWriteSent < System.currentTimeMillis() - config.exportInterval * 1000
|
||||
}
|
||||
|
||||
private fun conditionsForRemoteWrite() : Boolean {
|
||||
return deviceHasInternet() && ( timeHasPassed() || enoughSamplesScraped() )
|
||||
private fun conditionsForRemoteWrite(): Boolean {
|
||||
return deviceHasInternet() && (timeHasPassed() || enoughSamplesScraped())
|
||||
}
|
||||
|
||||
private fun enoughSamplesScraped() : Boolean {
|
||||
private fun enoughSamplesScraped(): Boolean {
|
||||
return storage.getLength() > config.maxSamplesPerExport
|
||||
}
|
||||
|
||||
private suspend fun senderManager(channel : Channel<Unit>){
|
||||
private suspend fun senderManager(channel: Channel<Unit>) {
|
||||
while (true) {
|
||||
if (storage.isEmpty()){
|
||||
if (storage.isEmpty()) {
|
||||
// channel is conflated, one receive is enough
|
||||
// suspend here until sending remote write is needed
|
||||
channel.receive()
|
||||
@ -157,37 +158,37 @@ class RemoteWriteSender(private val config: RemoteWriteConfiguration) {
|
||||
}
|
||||
|
||||
// entrypoint
|
||||
suspend fun start(){
|
||||
// conflated channel
|
||||
val channel = Channel<Unit>(1, onBufferOverflow = BufferOverflow.DROP_OLDEST)
|
||||
suspend fun start() {
|
||||
// conflated channel
|
||||
val channel = Channel<Unit>(1, onBufferOverflow = BufferOverflow.DROP_OLDEST)
|
||||
|
||||
client = HttpClient()
|
||||
try {
|
||||
runBlocking {
|
||||
launch {
|
||||
// check for outage in scrapes, save scrapes to storage
|
||||
scraper(channel)
|
||||
}
|
||||
launch {
|
||||
// send saved scrapes to remote write endpoint
|
||||
senderManager(channel)
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
withContext(NonCancellable){
|
||||
channel.close()
|
||||
client.close()
|
||||
Log.v(TAG, "Canceling Remote Write Sender")
|
||||
}
|
||||
}
|
||||
client = HttpClient()
|
||||
try {
|
||||
runBlocking {
|
||||
launch {
|
||||
// check for outage in scrapes, save scrapes to storage
|
||||
scraper(channel)
|
||||
}
|
||||
launch {
|
||||
// send saved scrapes to remote write endpoint
|
||||
senderManager(channel)
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
withContext(NonCancellable) {
|
||||
channel.close()
|
||||
client.close()
|
||||
Log.v(TAG, "Canceling Remote Write Sender")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fun countSuccessfulScrape(){
|
||||
fun countSuccessfulScrape() {
|
||||
Log.v(TAG, "Counting successful scrape")
|
||||
lastTimeRingBuffer.setLastTime(System.currentTimeMillis())
|
||||
}
|
||||
|
||||
private suspend fun sendRequestToRemoteWrite(body : ByteArray){
|
||||
private suspend fun sendRequestToRemoteWrite(body: ByteArray) {
|
||||
Log.v(TAG, "sending to prometheus remote write now")
|
||||
val response = client.post(config.remoteWriteEndpoint) {
|
||||
setBody(body)
|
||||
|
||||
@ -11,7 +11,7 @@ import java.util.Enumeration
|
||||
import java.util.LinkedList
|
||||
import java.util.Queue
|
||||
|
||||
private const val TAG : String = "REMOTE_WRITE_SENDER_STORAGE"
|
||||
private const val TAG: String = "REMOTE_WRITE_SENDER_STORAGE"
|
||||
|
||||
// This is a very primitive implementation, may require some optimization later
|
||||
//
|
||||
@ -24,21 +24,22 @@ typealias MetricsScrape = Enumeration<MetricFamilySamples>
|
||||
private typealias ConverterHashMap = HashMap<List<TimeSeriesLabel>, MutableList<TimeSeriesSample>>
|
||||
|
||||
private data class TimeSeriesLabel(
|
||||
val name : String,
|
||||
val value : String,
|
||||
){
|
||||
fun toProtobufLabel() : Label{
|
||||
val name: String,
|
||||
val value: String,
|
||||
) {
|
||||
fun toProtobufLabel(): Label {
|
||||
return Label.newBuilder()
|
||||
.setName(this.name)
|
||||
.setValue(this.value)
|
||||
.build()
|
||||
}
|
||||
}
|
||||
|
||||
private data class TimeSeriesSample(
|
||||
val timeStampMs : Long,
|
||||
val value : Double,
|
||||
){
|
||||
fun toProtobufSample() : Sample{
|
||||
val timeStampMs: Long,
|
||||
val value: Double,
|
||||
) {
|
||||
fun toProtobufSample(): Sample {
|
||||
return Sample.newBuilder()
|
||||
.setTimestamp(this.timeStampMs)
|
||||
.setValue(this.value)
|
||||
@ -47,26 +48,29 @@ private data class TimeSeriesSample(
|
||||
}
|
||||
|
||||
abstract class RemoteWriteSenderStorage {
|
||||
private val remoteWriteLabel : TimeSeriesLabel = TimeSeriesLabel(
|
||||
private val remoteWriteLabel: TimeSeriesLabel = TimeSeriesLabel(
|
||||
name = "backfill",
|
||||
value = "true",
|
||||
)
|
||||
|
||||
protected fun encodeWithSnappy(data: ByteArray): ByteArray {
|
||||
return Snappy.compress(data)
|
||||
}
|
||||
|
||||
private fun processLabels(sample : MetricFamilySamples.Sample,
|
||||
sampleName: String) : List<TimeSeriesLabel>{
|
||||
private fun processLabels(
|
||||
sample: MetricFamilySamples.Sample,
|
||||
sampleName: String
|
||||
): List<TimeSeriesLabel> {
|
||||
|
||||
val result : MutableList<TimeSeriesLabel> = mutableListOf()
|
||||
val result: MutableList<TimeSeriesLabel> = mutableListOf()
|
||||
|
||||
// labels are stored in parallel lists -> iterate over two lists at a time
|
||||
val sampleLabelNamesIterator = sample.labelNames.iterator()
|
||||
val sampleLabelValuesIterator = sample.labelNames.iterator()
|
||||
|
||||
while (sampleLabelNamesIterator.hasNext() && sampleLabelValuesIterator.hasNext()) {
|
||||
val labelName : String = sampleLabelNamesIterator.next()
|
||||
val labelValue : String = sampleLabelValuesIterator.next()
|
||||
val labelName: String = sampleLabelNamesIterator.next()
|
||||
val labelValue: String = sampleLabelValuesIterator.next()
|
||||
|
||||
val label = TimeSeriesLabel(
|
||||
name = labelName,
|
||||
@ -83,8 +87,8 @@ abstract class RemoteWriteSenderStorage {
|
||||
return result.toList()
|
||||
}
|
||||
|
||||
private fun getTimeSeriesSample(sample : MetricFamilySamples.Sample) : TimeSeriesSample{
|
||||
val timestampMs : Long = sample.timestampMs ?: System.currentTimeMillis()
|
||||
private fun getTimeSeriesSample(sample: MetricFamilySamples.Sample): TimeSeriesSample {
|
||||
val timestampMs: Long = sample.timestampMs ?: System.currentTimeMillis()
|
||||
|
||||
return TimeSeriesSample(
|
||||
value = sample.value,
|
||||
@ -93,18 +97,19 @@ abstract class RemoteWriteSenderStorage {
|
||||
}
|
||||
|
||||
private fun processTimeSeries(
|
||||
hashMap: ConverterHashMap, familySample : MetricFamilySamples){
|
||||
hashMap: ConverterHashMap, familySample: MetricFamilySamples
|
||||
) {
|
||||
|
||||
for ( sample in familySample.samples ) {
|
||||
val sampleName : String = sample.name
|
||||
val labels : List<TimeSeriesLabel> = processLabels(sample, sampleName)
|
||||
for (sample in familySample.samples) {
|
||||
val sampleName: String = sample.name
|
||||
val labels: List<TimeSeriesLabel> = processLabels(sample, sampleName)
|
||||
|
||||
val timeSeriesSample : TimeSeriesSample = getTimeSeriesSample(sample)
|
||||
val timeSeriesSample: TimeSeriesSample = getTimeSeriesSample(sample)
|
||||
|
||||
if (hashMap[labels] == null){
|
||||
if (hashMap[labels] == null) {
|
||||
// this time series does not yet exist
|
||||
hashMap[labels] = mutableListOf(timeSeriesSample)
|
||||
}else{
|
||||
} else {
|
||||
// this time series already exists
|
||||
hashMap[labels]!!.add(timeSeriesSample)
|
||||
}
|
||||
@ -112,25 +117,26 @@ abstract class RemoteWriteSenderStorage {
|
||||
}
|
||||
|
||||
private fun hashMapEntryToProtobufTimeSeries(
|
||||
labels : List<TimeSeriesLabel>, samples : MutableList<TimeSeriesSample>) : TimeSeries {
|
||||
labels: List<TimeSeriesLabel>, samples: MutableList<TimeSeriesSample>
|
||||
): TimeSeries {
|
||||
|
||||
val timeSeriesBuilder : TimeSeries.Builder = TimeSeries.newBuilder()
|
||||
val timeSeriesBuilder: TimeSeries.Builder = TimeSeries.newBuilder()
|
||||
|
||||
timeSeriesBuilder.addAllLabels(labels.map{
|
||||
timeSeriesBuilder.addAllLabels(labels.map {
|
||||
it.toProtobufLabel()
|
||||
})
|
||||
|
||||
timeSeriesBuilder.addAllSamples(samples.map{
|
||||
timeSeriesBuilder.addAllSamples(samples.map {
|
||||
it.toProtobufSample()
|
||||
})
|
||||
|
||||
return timeSeriesBuilder.build()
|
||||
}
|
||||
|
||||
private fun hashmapToProtobufWriteRequest(hashMap: ConverterHashMap) : WriteRequest{
|
||||
val writeRequestBuilder : WriteRequest.Builder = WriteRequest.newBuilder()
|
||||
private fun hashmapToProtobufWriteRequest(hashMap: ConverterHashMap): WriteRequest {
|
||||
val writeRequestBuilder: WriteRequest.Builder = WriteRequest.newBuilder()
|
||||
|
||||
for (entry in hashMap){
|
||||
for (entry in hashMap) {
|
||||
val timeSeries = hashMapEntryToProtobufTimeSeries(entry.key, entry.value)
|
||||
writeRequestBuilder.addTimeseries(timeSeries)
|
||||
}
|
||||
@ -138,20 +144,20 @@ abstract class RemoteWriteSenderStorage {
|
||||
return writeRequestBuilder.build()
|
||||
}
|
||||
|
||||
protected fun metricsScrapeListToProtobuf(input: List<MetricsScrape>) : WriteRequest {
|
||||
if(input.isEmpty()){
|
||||
protected fun metricsScrapeListToProtobuf(input: List<MetricsScrape>): WriteRequest {
|
||||
if (input.isEmpty()) {
|
||||
throw Exception("Input is empty!")
|
||||
}
|
||||
|
||||
val hashmap : ConverterHashMap = HashMap()
|
||||
val hashmap: ConverterHashMap = HashMap()
|
||||
|
||||
for ( metricsScrape in input ){
|
||||
for ( familySample in metricsScrape ) {
|
||||
for (metricsScrape in input) {
|
||||
for (familySample in metricsScrape) {
|
||||
processTimeSeries(hashmap, familySample)
|
||||
}
|
||||
}
|
||||
|
||||
val result : WriteRequest = hashmapToProtobufWriteRequest(hashmap)
|
||||
val result: WriteRequest = hashmapToProtobufWriteRequest(hashmap)
|
||||
|
||||
Log.v(TAG, result.timeseriesList.toString() + "<- protobuf")
|
||||
|
||||
@ -166,34 +172,34 @@ abstract class RemoteWriteSenderStorage {
|
||||
}
|
||||
|
||||
class RemoteWriteSenderSimpleMemoryStorage : RemoteWriteSenderStorage() {
|
||||
private val data : Queue<MetricsScrape> = LinkedList()
|
||||
private val data: Queue<MetricsScrape> = LinkedList()
|
||||
|
||||
override fun getScrapedSamplesCompressedProtobuf(howMany: Int): ByteArray {
|
||||
if (howMany < 1){
|
||||
if (howMany < 1) {
|
||||
throw IllegalArgumentException("howMany must be bigger than zero")
|
||||
}
|
||||
|
||||
val scrapedMetrics : MutableList<MetricsScrape> = mutableListOf()
|
||||
for (i in 1..howMany){
|
||||
val oneMetric : MetricsScrape? = data.poll()
|
||||
if(oneMetric == null){
|
||||
val scrapedMetrics: MutableList<MetricsScrape> = mutableListOf()
|
||||
for (i in 1..howMany) {
|
||||
val oneMetric: MetricsScrape? = data.poll()
|
||||
if (oneMetric == null) {
|
||||
break
|
||||
}else{
|
||||
} else {
|
||||
scrapedMetrics.add(oneMetric)
|
||||
}
|
||||
}
|
||||
|
||||
val writeRequest : WriteRequest = this.metricsScrapeListToProtobuf(scrapedMetrics.toList())
|
||||
val bytes : ByteArray = writeRequest.toByteArray()
|
||||
val writeRequest: WriteRequest = this.metricsScrapeListToProtobuf(scrapedMetrics.toList())
|
||||
val bytes: ByteArray = writeRequest.toByteArray()
|
||||
return this.encodeWithSnappy(bytes)
|
||||
}
|
||||
|
||||
override fun removeNumberOfScrapedSamples(number: Int) {
|
||||
if (number > 0){
|
||||
for (i in 1..number){
|
||||
if (number > 0) {
|
||||
for (i in 1..number) {
|
||||
data.remove()
|
||||
}
|
||||
}else{
|
||||
} else {
|
||||
throw IllegalArgumentException("number must by higher than 0")
|
||||
}
|
||||
}
|
||||
|
||||
@ -4,18 +4,18 @@ private const val TAG = "SCRAPE_RECORDER"
|
||||
|
||||
//TODO implement this thing
|
||||
// mutex with last scraped time
|
||||
class ScrapeRecorder{
|
||||
class ScrapeRecorder {
|
||||
|
||||
//TODO mutex variable if mode is {pushprox / prometheus server} or {remote write}
|
||||
//TODO go back to mode {pushprox / prometheus server} only after N succesfull scrapes and no failures
|
||||
|
||||
fun countSuccesfullScrape(){
|
||||
fun countSuccesfullScrape() {
|
||||
//TODO implement this thing
|
||||
// write to mutex that scrape has happend at this current time
|
||||
// set timer to 2 x remote_write_scrape_interval seconds to check if next scrape has happened
|
||||
}
|
||||
|
||||
private fun onTimerTick(){
|
||||
private fun onTimerTick() {
|
||||
//TODO implement this
|
||||
// check if other scrape has happened
|
||||
// if no scrape happened, go to mode {remote write}
|
||||
|
||||
Reference in New Issue
Block a user