pushprox refactored

This commit is contained in:
Martin Ptáček
2023-06-01 18:59:11 +02:00
parent 27c62e0731
commit c0cdae89ec
4 changed files with 120 additions and 79 deletions

View File

@ -99,15 +99,9 @@ class PromViewModel(): ViewModel() {
when(_uiState.value.exporterState) { when(_uiState.value.exporterState) {
ExporterState.Running -> { ExporterState.Running -> {
stopWorker() stopWorker()
_uiState.update { current ->
current.copy(exporterState = ExporterState.NotRunning)
}
} }
ExporterState.NotRunning -> { ExporterState.NotRunning -> {
startWorker() startWorker()
_uiState.update { current ->
current.copy(exporterState = ExporterState.Running)
}
} }
} }
} }
@ -125,7 +119,13 @@ class PromViewModel(): ViewModel() {
} }
} }
private fun validatePromConfiguration() : Boolean{
//TODO implement this, on missing fields, e.g. not valid, display alert dialog in UI
return true
}
private fun startWorker(){ private fun startWorker(){
if (validatePromConfiguration()){
Log.v(TAG, "Enqueuing work") Log.v(TAG, "Enqueuing work")
val workManagerInstance = WorkManager.getInstance(getContext()) val workManagerInstance = WorkManager.getInstance(getContext())
@ -150,11 +150,22 @@ class PromViewModel(): ViewModel() {
ExistingWorkPolicy.KEEP, ExistingWorkPolicy.KEEP,
workerRequest, workerRequest,
).enqueue() ).enqueue()
// set UI state
_uiState.update { current ->
current.copy(exporterState = ExporterState.Running)
}
}
} }
private fun stopWorker(){ private fun stopWorker(){
val workerManagerInstance = WorkManager.getInstance(getContext()) val workerManagerInstance = WorkManager.getInstance(getContext())
workerManagerInstance.cancelUniqueWork(PROM_UNIQUE_WORK) workerManagerInstance.cancelUniqueWork(PROM_UNIQUE_WORK)
// update UI state
_uiState.update { current ->
current.copy(exporterState = ExporterState.NotRunning)
}
} }
fun updatePromConfig(part : UpdatePromConfig, value : Any){ fun updatePromConfig(part : UpdatePromConfig, value : Any){

View File

@ -0,0 +1,44 @@
package com.birdthedeveloper.prometheus.android.prometheus.android.exporter.worker
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.delay
import kotlin.math.pow
class ExponentialBackoff{
companion object{
private const val multiplier : Double = 2.0
private const val initialDelay = 3.0 // seconds
suspend fun runWithBackoff(function : suspend () -> Unit, onException: () -> Unit){
var successfull : Boolean = false
var currentDelay = initialDelay
var currentExpIndex = -1
while(!successfull){
try {
function()
successfull = true
}catch (e : CancellationException){
successfull = true
}
catch (e : Exception){
// check for suppressed exceptions
for(exception in e.suppressed){
if(exception is CancellationException){
successfull = true
}
}
onException()
// calculate new delay
currentExpIndex++
currentDelay = initialDelay + multiplier.pow(currentExpIndex)
}
delay(currentDelay.toLong())
}
}
}
}

View File

@ -27,9 +27,10 @@ class PromWorker(
parameters : WorkerParameters, parameters : WorkerParameters,
) : CoroutineWorker(context, parameters) { ) : CoroutineWorker(context, parameters) {
private val collectorRegistry = CollectorRegistry()
private val metricsEngine : MetricsEngine = MetricsEngine(context) private val metricsEngine : MetricsEngine = MetricsEngine(context)
private val pushProxClient = PushProxClient(::performScrape)
private lateinit var androidCustomExporter : AndroidCustomExporter private lateinit var androidCustomExporter : AndroidCustomExporter
private lateinit var pushProxClient : PushProxClient
//TODO foreground notification //TODO foreground notification
private val notificationManager = private val notificationManager =
@ -44,7 +45,7 @@ class PromWorker(
private fun initializeWork(config : PromConfiguration){ private fun initializeWork(config : PromConfiguration){
// initialize metrics // initialize metrics
androidCustomExporter = AndroidCustomExporter(metricsEngine).register() androidCustomExporter = AndroidCustomExporter(metricsEngine).register(collectorRegistry)
} }
private suspend fun startServices(config : PromConfiguration){ private suspend fun startServices(config : PromConfiguration){
@ -54,13 +55,20 @@ class PromWorker(
PrometheusServer.start( PrometheusServer.start(
PrometheusServerConfig(config.prometheusServerPort, ::performScrape), PrometheusServerConfig(config.prometheusServerPort, ::performScrape),
) )
Log.v(TAG, "Prometheus server started.")
} }
} }
if(config.pushproxEnabled){ if(config.pushproxEnabled){
pushProxClient = PushProxClient(
PushProxConfig(
pushProxUrl = config.pushproxProxyUrl,
performScrape = ::performScrape,
pushProxFqdn = config.pushproxFqdn,
registry = collectorRegistry,
)
)
launch { launch {
pushProxClient.start()
} }
} }
@ -81,7 +89,6 @@ class PromWorker(
initializeWork(inputConfiguration) initializeWork(inputConfiguration)
startServices(inputConfiguration) startServices(inputConfiguration)
//TODO implement this asap
return Result.success() return Result.success()
} }

View File

@ -20,6 +20,7 @@ data class PushProxConfig(
val pushProxUrl : String, val pushProxUrl : String,
val pushProxFqdn : String, val pushProxFqdn : String,
val registry : CollectorRegistry, val registry : CollectorRegistry,
val performScrape : () -> String,
) )
/** /**
@ -43,10 +44,10 @@ private class PushProxCounter(registry: CollectorRegistry) {
fun scrapeError(){ scrapeErrorCounter.inc()} fun scrapeError(){ scrapeErrorCounter.inc()}
fun pushError(){ pushErrorCounter.inc() } fun pushError(){ pushErrorCounter.inc() }
fun pollError(){ pollErrorCounter.inc() } //TODO use this thing fun pollError(){ pollErrorCounter.inc() }
} }
// Error in parsing HTTP header "Id" from HTTP request from Prometheus //TODO wtf // Error in parsing HTTP header "Id" from HTTP request from Prometheus
class PushProxIdParseException(message: String) : Exception(message) class PushProxIdParseException(message: String) : Exception(message)
// Context object for pushprox internal functions to avoid global variables // Context object for pushprox internal functions to avoid global variables
@ -58,17 +59,17 @@ data class PushProxContext(
) )
// This is a stripped down kotlin implementation of github.com/prometheus-community/PushProx client // This is a stripped down kotlin implementation of github.com/prometheus-community/PushProx client
class PushProxClient(config: PushProxConfig) { class PushProxClient(private val pushProxConfig: PushProxConfig) {
private val counters : PushProxCounter = PushProxCounter(config.registry) private val counters : PushProxCounter = PushProxCounter(pushProxConfig.registry)
// Use this function to start exporting metrics to pushprox in the background // Use this function to start exporting metrics to pushprox in the background
suspend fun start(config: PushProxConfig) { suspend fun start() {
Log.v(TAG, "Starting pushprox client") Log.v(TAG, "Starting pushprox client")
var client : HttpClient? = null var client : HttpClient? = null
try { try {
client = HttpClient() client = HttpClient()
val context : PushProxContext = getPushProxContext(client, config) val context : PushProxContext = getPushProxContext(client)
loop(context) loop(context)
}finally { }finally {
withContext(NonCancellable){ withContext(NonCancellable){
@ -78,9 +79,8 @@ class PushProxClient(config: PushProxConfig) {
} }
} }
private fun getPushProxContext(client : HttpClient) : PushProxContext {
private fun getPushProxContext(client : HttpClient, config : PushProxConfig) : PushProxContext { var modifiedProxyURL = pushProxConfig.pushProxUrl.trim('/')
var modifiedProxyURL = config.pushProxUrl.trim('/')
if( if(
!modifiedProxyURL.startsWith("http://") && !modifiedProxyURL.startsWith("http://") &&
@ -96,23 +96,19 @@ class PushProxClient(config: PushProxConfig) {
client, client,
pollURL, pollURL,
pushURL, pushURL,
config.pushProxFqdn, pushProxConfig.pushProxFqdn,
) )
} }
//TODO refactor this function // Continuously poll from pushprox gateway
// Continuous poll from android phone to pushprox gateway
private suspend fun doPoll(context : PushProxContext){ private suspend fun doPoll(context : PushProxContext){
log("poll", "polling now")
val response : HttpResponse = context.client.post(context.pollUrl){ val response : HttpResponse = context.client.post(context.pollUrl){
setBody(context.fqdn) setBody(context.fqdn)
} }
log("here", "here")
val responseBody: String = response.body<String>() val responseBody: String = response.body<String>()
doPush(context, responseBody) doPush(context, responseBody)
} }
//TODO refactor this function
// get value of HTTP header "Id" from response body // get value of HTTP header "Id" from response body
private fun getIdFromResponseBody(responseBody: String) : String { private fun getIdFromResponseBody(responseBody: String) : String {
@ -124,7 +120,6 @@ class PushProxClient(config: PushProxConfig) {
return id return id
} }
//TODO refactor this function
private fun composeRequestBody(scrapedMetrics: String, id : String) : String { private fun composeRequestBody(scrapedMetrics: String, id : String) : String {
val httpHeaders = "HTTP/1.1 200 OK\r\n" + val httpHeaders = "HTTP/1.1 200 OK\r\n" +
"Content-Type: text/plain; version=0.0.4; charset=utf-8\r\n" + "Content-Type: text/plain; version=0.0.4; charset=utf-8\r\n" +
@ -137,55 +132,39 @@ class PushProxClient(config: PushProxConfig) {
// Parameter responseBody: response body of /poll request // Parameter responseBody: response body of /poll request
private suspend fun doPush(context : PushProxContext, pollResponseBody : String) { private suspend fun doPush(context : PushProxContext, pollResponseBody : String) {
// perform scrape // perform scrape
lateinit var scrapedMetrics : String val scrapedMetrics : String = try {
try { pushProxConfig.performScrape()
scrapedMetrics = performScrape() }catch (e : Exception){
}catch(e : Exception){ Log.v(TAG, "Scrape exception ${e.toString()}")
counters.scrapeError() counters.scrapeError()
log("scrape exception", e.toString()) ""
return
} }
// push metrics to pushprox
try{ try{
val scrapeId : String = getIdFromResponseBody(pollResponseBody) val scrapeId : String = getIdFromResponseBody(pollResponseBody)
val pushResponseBody: String = composeRequestBody(scrapedMetrics, scrapeId) val pushRequestBody: String = composeRequestBody(scrapedMetrics, scrapeId)
val response : HttpResponse = context.client.request(context.pushUrl) { context.client.request(context.pushUrl) {
method = HttpMethod.Post method = HttpMethod.Post
setBody(pushResponseBody) setBody(pushRequestBody)
} }
}catch(e : Exception){ }catch(e : Exception){
counters.pushError() counters.pushError()
log("push exception", e.toString()) Log.v(TAG,"Push exception ${e.toString()}")
return return
} }
} }
//TODO migrate to work manager
private suspend fun loop(context : PushProxContext) { private suspend fun loop(context : PushProxContext) {
while (true) { while (true) {
Log.v(TAG, "PushProxClient main loop start") Log.v(TAG, "PushProxClient main loop start")
// register poll error using try-catch block
//TODO backoff strategy ExponentialBackoff.runWithBackoff(
//TODO asap function = { doPoll(context) },
// var result = context.backoff.withRetries { onException = { counters.pollError() }
// try { )
// doPoll(context)
// }catch(e : CancellationException){
// shouldContinue = false
// }
// catch (e: Exception) {
// for(exception in e.suppressed){
// if(exception is CancellationException){
// shouldContinue = false
// }
// }
// log("exception encountered!", e.toString())
// counters.pollError()
// throw e
// }
// }
Log.v(TAG,"PushProxClient main loop end") Log.v(TAG,"PushProxClient main loop end")
} }
} }