mozilla
/
glean
зеркало из https://github.com/mozilla/glean.git
1
0
Форкнуть 0
Код Задачи Пакеты Проекты Релизы Вики Активность

[Kotlin, Rust] Enable upload process through new API

This commit is contained in:
Jan-Erik Rediger 2021-11-11 18:01:01 +01:00
Родитель 7cc543ed1b
Коммит 9c843ec279
20 изменённых файлов: 312 добавлений и 543 удалений
Показать статистику Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

53
glean-core/android/src/main/java/mozilla/telemetry/glean/Glean.kt
Просмотреть файл

@ -62,7 +62,16 @@ internal class OnGleanEventsImpl(
override fun startMetricsPingScheduler() {
glean.metricsPingScheduler = MetricsPingScheduler(glean.applicationContext, glean.buildInfo)
glean.metricsPingScheduler.schedule()
glean.metricsPingScheduler?.schedule()
}
override fun cancelUploads() {
// Cancel any pending workers here so that we don't accidentally upload or
// collect data after the upload has been disabled.
glean.metricsPingScheduler?.cancel()
// Cancel any pending workers here so that we don't accidentally upload
// data after the upload has been disabled.
PingUploadWorker.cancel(glean.applicationContext)
}
}
@ -106,7 +115,7 @@ open class GleanInternalAPI internal constructor () {
// This object holds data related to any persistent information about the metrics ping,
// such as the last time it was sent and the store name
internal lateinit var metricsPingScheduler: MetricsPingScheduler
internal var metricsPingScheduler: MetricsPingScheduler? = null
// This is used to cache the process state and is used by the function `isMainProcess()`
@VisibleForTesting(otherwise = VisibleForTesting.PRIVATE)
@ -119,6 +128,10 @@ open class GleanInternalAPI internal constructor () {
// Store the build information provided by the application.
internal lateinit var buildInfo: BuildInfo
init {
gleanEnableLogging()
}
/**
* Initialize the Glean SDK.
*
@ -179,9 +192,6 @@ open class GleanInternalAPI internal constructor () {
this.httpClient = BaseUploader(configuration.httpClient)
this.gleanDataDir = File(applicationContext.applicationInfo.dataDir, GLEAN_DATA_DIR)
Log.e(LOG_TAG, "Glean. enable logging.")
gleanEnableLogging()
val cfg = InternalConfiguration(
dataPath = gleanDataDir.path,
applicationId = applicationContext.packageName,
@ -323,7 +333,8 @@ open class GleanInternalAPI internal constructor () {
// time the app goes to background.
gleanHandleClientActive()
GleanValidation.foregroundCount.add(1)
// TODO: Re-enable when Counter metric is working
//GleanValidation.foregroundCount.add(1)
}
/**
@ -356,32 +367,6 @@ open class GleanInternalAPI internal constructor () {
gleanSubmitPingByName(pingName, reason)
}
/**
* Collect and submit a ping (by its name) for eventual upload, synchronously.
*
* The ping will be looked up in the known instances of [PingType]. If the
* ping isn't known, an error is logged and the ping isn't queued for uploading.
*
* The ping content is assembled as soon as possible, but upload is not
* guaranteed to happen immediately, as that depends on the upload
* policies.
*
* If the ping currently contains no content, it will not be assembled and
* queued for sending, unless explicitly specified otherwise in the registry
* file.
*
* @param pingName Name of the ping to submit.
* @param reason The reason the ping is being submitted.
*/
internal fun submitPingByNameSync(pingName: String, reason: String? = null) {
if (!isInitialized()) {
Log.e(LOG_TAG, "Glean must be initialized before submitting pings.")
return
}
gleanSubmitPingByNameSync(pingName, reason)
}
/**
* Set a tag to be applied to headers when uploading pings for debug view.
*
@ -453,6 +438,10 @@ open class GleanInternalAPI internal constructor () {
isMainProcess = null
// Resetting MPS and uploader
metricsPingScheduler?.cancel()
PingUploadWorker.cancel(context)
// Init Glean.
Glean.testDestroyGleanHandle(clearStores)
// Always log pings for tests

6
glean-core/android/src/main/java/mozilla/telemetry/glean/net/HttpURLConnectionUploader.kt
Просмотреть файл

@ -64,16 +64,16 @@ class HttpURLConnectionUploader : PingUploader {
// Finally upload.
val statusCode = doUpload(connection, data)
return HttpResponse(statusCode)
return HttpStatus(statusCode)
} catch (e: MalformedURLException) {
// There's nothing we can do to recover from this here. So let's just log an error and
// notify the service that this job has been completed - even though we didn't upload
// anything to the server.
Log.e(LOG_TAG, "Could not upload telemetry due to malformed URL", e)
return UnrecoverableFailure
return UnrecoverableFailure(0)
} catch (e: IOException) {
Log.w(LOG_TAG, "IOException while uploading ping", e)
return RecoverableFailure
return RecoverableFailure(0)
} finally {
connection?.disconnect()
}

51
glean-core/android/src/main/java/mozilla/telemetry/glean/net/PingUploader.kt
Просмотреть файл

@ -4,60 +4,11 @@
package mozilla.telemetry.glean.net
import mozilla.telemetry.glean.rust.Constants
/**
* Store a list of headers as a String to String [Pair], with the first entry
* being the header name and the second its value.
*/
typealias HeadersList = List<Pair<String, String>>
/**
* The result of the ping upload.
*
* See below for the different possible cases.
*/
sealed class UploadResult {
open fun toFfi(): Int {
return Constants.UPLOAD_RESULT_UNRECOVERABLE
}
}
/**
* A HTTP response code.
*
* This can still indicate an error, depending on the status code.
*/
data class HttpResponse(val statusCode: Int) : UploadResult() {
override fun toFfi(): Int {
return Constants.UPLOAD_RESULT_HTTP_STATUS or statusCode
}
}
/**
* An unrecoverable upload failure.
*
* A possible cause might be a malformed URL.
* The ping data is removed afterwards.
*/
object UnrecoverableFailure : UploadResult() {
override fun toFfi(): Int {
return Constants.UPLOAD_RESULT_UNRECOVERABLE
}
}
/**
* A recoverable failure.
*
* During upload something went wrong,
* e.g. the network connection failed.
* The upload should be retried at a later time.
*/
object RecoverableFailure : UploadResult() {
override fun toFfi(): Int {
return Constants.UPLOAD_RESULT_RECOVERABLE
}
}
typealias HeadersList = Map<String, String>
/**
* The interface defining how to send pings.

134
glean-core/android/src/main/java/mozilla/telemetry/glean/net/Upload.kt
Просмотреть файл

@ -10,132 +10,10 @@ import org.json.JSONException
import com.sun.jna.Structure
import com.sun.jna.Pointer
import com.sun.jna.Union
import mozilla.telemetry.glean.rust.getRustString
import mozilla.telemetry.glean.rust.RustBuffer
// Rust represents the upload task as an Enum
// and to go through the FFI that gets transformed into a tagged union.
// Each variant is represented as an 8-bit unsigned integer.
//
// This *MUST* have the same order as the variants in `glean-core/ffi/src/upload.rs`.
enum class UploadTaskTag {
Upload,
Wait,
Done
}
@Structure.FieldOrder("tag", "documentId", "path", "body", "headers")
internal class UploadBody(
// NOTE: We need to provide defaults here, so that JNA can create this object.
@JvmField val tag: Byte = UploadTaskTag.Upload.ordinal.toByte(),
@JvmField val documentId: Pointer? = null,
@JvmField val path: Pointer? = null,
@JvmField var body: RustBuffer = RustBuffer(),
@JvmField val headers: Pointer? = null
) : Structure() {
fun toPingRequest(): PingRequest {
return PingRequest(
this.documentId!!.getRustString(),
this.path!!.getRustString(),
this.body.getByteArray(),
this.headers!!.getRustString()
)
}
}
@Structure.FieldOrder("tag", "time")
internal class WaitBody(
@JvmField var tag: Byte = UploadTaskTag.Wait.ordinal.toByte(),
@JvmField var time: Long = 60_000
) : Structure()
internal open class FfiPingUploadTask(
// NOTE: We need to provide defaults here, so that JNA can create this object.
@JvmField var tag: Byte = UploadTaskTag.Done.ordinal.toByte(),
@JvmField var upload: UploadBody = UploadBody(),
@JvmField var wait: WaitBody = WaitBody()
) : Union() {
class ByReference : FfiPingUploadTask(), Structure.ByReference
init {
// Initialize to be the `tag`-only variant
setType("tag")
}
fun toPingUploadTask(): PingUploadTask {
return when (this.tag.toInt()) {
UploadTaskTag.Wait.ordinal -> {
this.readField("wait")
PingUploadTask.Wait(this.wait.time)
}
UploadTaskTag.Upload.ordinal -> {
this.readField("upload")
PingUploadTask.Upload(this.upload.toPingRequest())
}
else -> PingUploadTask.Done
}
}
}
/**
* Represents a request to upload a ping.
*/
internal class PingRequest(
private val documentId: String,
val path: String,
val body: ByteArray,
headers: String
) {
val headers: HeadersList = headersFromJSONString(headers)
private fun headersFromJSONString(str: String): HeadersList {
val headers: MutableList<Pair<String, String>> = mutableListOf()
try {
val jsonHeaders = JSONObject(str)
for (key in jsonHeaders.keys()) {
headers.add(Pair(key, jsonHeaders.get(key).toString()))
}
} catch (e: JSONException) {
// This JSON is created on the Rust side right before sending them over the FFI,
// it's very unlikely that we get an exception here
// unless there is some sort of memory corruption.
Log.e(LOG_TAG, "Error while parsing headers for ping $documentId")
}
return headers
}
companion object {
private const val LOG_TAG: String = "glean/Upload"
}
}
/**
* When asking for the next ping request to upload,
* the requester may receive one out of three possible tasks.
*/
internal sealed class PingUploadTask {
/**
* A PingRequest popped from the front of the queue.
*/
class Upload(val request: PingRequest) : PingUploadTask()
/**
* A flag signaling that the pending pings directories are not done being processed,
* thus the requester should wait and come back later.
*/
data class Wait(val time: Long) : PingUploadTask()
/**
* A flag signaling that requester doesn't need to request any more upload tasks at this moment.
*
* There are two possibilities for this scenario:
* * Pending pings queue is empty, no more pings to request;
* * Requester has reported more max recoverable upload failures on the same uploading_window[1]
* and should stop requesting at this moment.
*
* [1]: An "uploading window" starts when a requester gets a new `PingUploadTask::Upload(PingRequest)`
* response and finishes when they finally get a `PingUploadTask::Done` or `PingUploadTask::Wait` response.
*/
object Done : PingUploadTask()
}
typealias PingUploadTask = mozilla.telemetry.glean.internal.PingUploadTask
typealias PingRequest = mozilla.telemetry.glean.internal.PingRequest
typealias UploadResult = mozilla.telemetry.glean.internal.UploadResult
typealias HttpStatus = mozilla.telemetry.glean.internal.UploadResult.HttpStatus
typealias UnrecoverableFailure = mozilla.telemetry.glean.internal.UploadResult.UnrecoverableFailure
typealias RecoverableFailure = mozilla.telemetry.glean.internal.UploadResult.RecoverableFailure

6
glean-core/android/src/main/java/mozilla/telemetry/glean/private/PingType.kt
Просмотреть файл

@ -4,12 +4,8 @@
package mozilla.telemetry.glean.private
import com.sun.jna.StringArray
import androidx.annotation.VisibleForTesting
import mozilla.telemetry.glean.Glean
import mozilla.telemetry.glean.Dispatchers
import mozilla.telemetry.glean.rust.LibGleanFFI
import mozilla.telemetry.glean.rust.toByte
import mozilla.telemetry.glean.internal.PingType as GleanPingType
/**
@ -65,8 +61,6 @@ class PingType<ReasonCodesEnum : Enum<ReasonCodesEnum>> (
@VisibleForTesting(otherwise = VisibleForTesting.NONE)
@Synchronized
fun testBeforeNextSubmit(cb: (ReasonCodesEnum?) -> Unit) {
@Suppress("EXPERIMENTAL_API_USAGE")
Dispatchers.API.assertInTestingMode()
this.testCallback = cb
}

30
glean-core/android/src/main/java/mozilla/telemetry/glean/rust/LibGleanFFI.kt
Просмотреть файл

@ -13,7 +13,6 @@ import com.sun.jna.Pointer
import com.sun.jna.StringArray
import java.lang.reflect.Proxy
import mozilla.telemetry.glean.config.FfiConfiguration
import mozilla.telemetry.glean.net.FfiPingUploadTask
// Turn a boolean into its Byte (u8) representation
internal fun Boolean.toByte(): Byte = if (this) 1 else 0
@ -21,31 +20,6 @@ internal fun Boolean.toByte(): Byte = if (this) 1 else 0
// Turn a Byte into a boolean where zero is false and non-zero is true
internal fun Byte.toBoolean(): Boolean = this != 0.toByte()
/**
* Result values of attempted ping uploads encoded for FFI use.
* They are defined in `glean-core/src/upload/result.rs` and re-defined for use in Kotlin here.
*
* NOTE:
* THEY MUST BE THE SAME ACROSS BOTH FILES!
*/
class Constants {
private constructor() {
// hiding the constructor.
// intentionally left empty otherwise.
}
companion object {
// A recoverable error.
const val UPLOAD_RESULT_RECOVERABLE: Int = 0x1
// An unrecoverable error.
const val UPLOAD_RESULT_UNRECOVERABLE: Int = 0x2
// A HTTP response code.
const val UPLOAD_RESULT_HTTP_STATUS: Int = 0x8000
}
}
/**
* Helper to read a null terminated String out of the Pointer and free it.
*
@ -647,10 +621,6 @@ internal interface LibGleanFFI : Library {
storage_name: String
): Int
fun glean_get_upload_task(task: FfiPingUploadTask.ByReference)
fun glean_process_ping_upload_response(task: FfiPingUploadTask.ByReference, status: Int)
fun glean_set_debug_view_tag(value: String): Byte
fun glean_set_log_pings(value: Byte)

53
glean-core/android/src/main/java/mozilla/telemetry/glean/rust/RustBuffer.kt
Просмотреть файл

@ -1,53 +0,0 @@
/* This Source Code Form is subject to the terms of the Mozilla Public
* License, v. 2.0. If a copy of the MPL was not distributed with this
* file, You can obtain one at http://mozilla.org/MPL/2.0/. */
package mozilla.telemetry.glean.rust
import com.sun.jna.Pointer
import com.sun.jna.Structure
/**
* This is a mapping for the `glean_core::byte_buffer::ByteBuffer` struct.
* It's a copy of the `support.native.RustBuffer` class in application-services,
* with the small changes that the length field is a (32-bit) Int instead.
*
* The name differs for two reasons.
*
* 1. To that the memory this type manages is allocated from rust code,
* and must subsequently be freed by rust code.
*
* 2. To avoid confusion with java's nio ByteBuffer, which we use for
* passing data *to* Rust without incurring additional copies.
*
* # Caveats:
*
* 1. It is for receiving data *FROM* Rust, and not the other direction.
* RustBuffer doesn't expose a way to inspect its contents from Rust.
* See `docs/howtos/passing-protobuf-data-over-ffi.md` for how to do
* this instead.
*
* 2. A `RustBuffer` passed into kotlin code must be freed by kotlin
* code *after* the protobuf message is completely deserialized.
*
* The rust code must expose a destructor for this purpose,
* and it should be called in the finally block after the data
* is read from the `CodedInputStream` (and not before).
*
* 3. You almost always should use `RustBuffer.ByValue` instead
* of `RustBuffer`. E.g.
* `fun mylib_get_stuff(some: X, args: Y): RustBuffer.ByValue`
* for the function returning the RustBuffer, and
* `fun mylib_destroy_bytebuffer(bb: RustBuffer.ByValue)`.
*/
@Structure.FieldOrder("len", "data")
open class RustBuffer : Structure() {
@JvmField var len: Int = 0
@JvmField var data: Pointer? = null
fun getByteArray(): ByteArray {
return this.data!!.getByteArray(0, this.len)
}
class ByValue : RustBuffer(), Structure.ByValue
}

8
glean-core/android/src/main/java/mozilla/telemetry/glean/scheduler/MetricsPingScheduler.kt
Просмотреть файл

@ -11,6 +11,7 @@ import android.text.format.DateUtils
import android.util.Log
import mozilla.telemetry.glean.Dispatchers
import mozilla.telemetry.glean.Glean
import mozilla.telemetry.glean.internal.gleanSubmitPingByNameSync
import mozilla.telemetry.glean.BuildInfo
import mozilla.telemetry.glean.GleanMetrics.Pings
import mozilla.telemetry.glean.utils.getISOTimeString
@ -49,6 +50,7 @@ internal class MetricsPingScheduler(
}
init {
Log.i(LOG_TAG, "New MetricsPingSched")
// In testing mode, set the "last seen version" as the same as this one.
// Otherwise, all we will ever send is pings for the "upgrade" reason.
@Suppress("EXPERIMENTAL_API_USAGE")
@ -105,7 +107,7 @@ internal class MetricsPingScheduler(
val millisUntilNextDueTime = getMillisecondsUntilDueTime(sendTheNextCalendarDay, now)
Log.d(LOG_TAG, "Scheduling the 'metrics' ping in ${millisUntilNextDueTime}ms")
// Cancel any existing scheduled work. Does not actually cancel a
// Cancel any existing scheduled work. Does not actually ancel a
// currently-running task.
cancel()
@ -211,6 +213,7 @@ internal class MetricsPingScheduler(
* collection.
*/
fun schedule() {
Log.i(LOG_TAG, "MetricsPingSched.schedule")
val now = getCalendarInstance()
// If the version of the app is different from the last time we ran the app,
@ -302,7 +305,8 @@ internal class MetricsPingScheduler(
//
// * Do not change this line without checking what it implies for the above wall
// of text. *
Glean.submitPingByNameSync("metrics", reasonString)
gleanSubmitPingByNameSync("metrics", reasonString)
// The initialization process will take care of triggering the uploader here.
} else {
Pings.metrics.submit(reason)
}

21
glean-core/android/src/main/java/mozilla/telemetry/glean/scheduler/PingUploadWorker.kt
Просмотреть файл

@ -15,11 +15,11 @@ import androidx.work.OneTimeWorkRequestBuilder
import androidx.work.WorkManager
import androidx.work.Worker
import androidx.work.WorkerParameters
import mozilla.telemetry.glean.rust.LibGleanFFI
import mozilla.telemetry.glean.internal.gleanGetUploadTask
import mozilla.telemetry.glean.internal.gleanProcessPingUploadResponse
import mozilla.telemetry.glean.internal.PingUploadTask
import mozilla.telemetry.glean.Glean
import mozilla.telemetry.glean.net.FfiPingUploadTask
import mozilla.telemetry.glean.utils.testFlushWorkManagerJob
import mozilla.telemetry.glean.net.PingUploadTask
/**
* Build the constraints around which the worker can be run, such as whether network
@ -97,27 +97,24 @@ class PingUploadWorker(context: Context, params: WorkerParameters) : Worker(cont
@Suppress("ReturnCount")
override fun doWork(): Result {
do {
// Create a slot of memory for the task: glean-core will write data into
// the allocated memory.
val incomingTask = FfiPingUploadTask.ByReference()
LibGleanFFI.INSTANCE.glean_get_upload_task(incomingTask)
when (val action = incomingTask.toPingUploadTask()) {
when (val action = gleanGetUploadTask()) {
is PingUploadTask.Upload -> {
// Upload the ping request.
// If the status is `null` there was some kind of unrecoverable error
// so we return a known unrecoverable error status code
// which will ensure this gets treated as such.
val body = action.request.body.toUByteArray().asByteArray()
val result = Glean.httpClient.doUpload(
action.request.path,
action.request.body,
body,
action.request.headers,
Glean.configuration
).toFfi()
)
// Process the upload response
LibGleanFFI.INSTANCE.glean_process_ping_upload_response(incomingTask, result)
gleanProcessPingUploadResponse(action.request.documentId, result)
}
is PingUploadTask.Wait -> SystemClock.sleep(action.time)
is PingUploadTask.Wait -> SystemClock.sleep(action.time.toLong())
is PingUploadTask.Done -> return Result.success()
}
} while (true)

99
glean-core/android/src/test/java/mozilla/telemetry/glean/FfiUploadTest.kt
Просмотреть файл

@ -1,99 +0,0 @@
/* This Source Code Form is subject to the terms of the Mozilla Public
* License, v. 2.0. If a copy of the MPL was not distributed with this
* file, You can obtain one at http://mozilla.org/MPL/2.0/. */
package mozilla.telemetry.glean
import android.content.Context
import androidx.test.core.app.ApplicationProvider
import androidx.test.ext.junit.runners.AndroidJUnit4
import kotlinx.coroutines.ObsoleteCoroutinesApi
import mozilla.telemetry.glean.private.NoReasonCodes
import mozilla.telemetry.glean.private.PingType
import mozilla.telemetry.glean.rust.LibGleanFFI
import mozilla.telemetry.glean.testing.GleanTestRule
import mozilla.telemetry.glean.net.FfiPingUploadTask
import mozilla.telemetry.glean.net.HttpResponse
import mozilla.telemetry.glean.net.PingUploadTask
import org.junit.After
import org.junit.Assert.assertTrue
import org.junit.Rule
import org.junit.Test
import org.junit.runner.RunWith
@ObsoleteCoroutinesApi
@RunWith(AndroidJUnit4::class)
class FfiUploadTest {
private val context: Context
get() = ApplicationProvider.getApplicationContext()
@get:Rule
val gleanRule = GleanTestRule(context)
@After
fun resetGlobalState() {
Glean.setUploadEnabled(true)
}
/**
* This test checks that the rate limiting works as expected:
*
* 1. 16 pings are submitted. The rate limit is 15 pings per minute, so we are one over.
* 2. The first 15 pings are send without delay.
* 3. On requesting the task for the 16th ping the upload manager asks the ping uploader
* to wait a specified amount of time.
* This time is less than a minute.
*
* Note:
* * This test does not wait for the full minute to expire to then get the final upload task.
* * We need to test the FFI boundary,
* which unfortunately requires to duplicate code that lives in the `PingUploadWorker`.
*/
@Test
fun `rate limiting instructs the uploader to wait shortly`() {
delayMetricsPing(context)
val server = getMockWebServer()
resetGlean(context, Glean.configuration.copy(
serverEndpoint = "http://" + server.hostName + ":" + server.port
))
val customPing = PingType<NoReasonCodes>(
name = "custom_ping",
includeClientId = true,
sendIfEmpty = true,
reasonCodes = listOf()
)
// Submit pings until the rate limit
for (i in 1..15) {
customPing.submit()
}
// Send exactly one more
customPing.submit()
// Expect to upload the first 15 ones.
for (i in 1..15) {
val incomingTask = FfiPingUploadTask.ByReference()
LibGleanFFI.INSTANCE.glean_get_upload_task(incomingTask)
val action = incomingTask.toPingUploadTask()
assertTrue("Task $i, expected Upload, was: ${action::class.qualifiedName}", action is PingUploadTask.Upload)
// Mark as uploaded.
val result = HttpResponse(200)
LibGleanFFI.INSTANCE.glean_process_ping_upload_response(incomingTask, result.toFfi())
}
// Task 16 is throttled, so the uploader needs to wait
val incomingTask = FfiPingUploadTask.ByReference()
LibGleanFFI.INSTANCE.glean_get_upload_task(incomingTask)
val action = incomingTask.toPingUploadTask()
assertTrue("Next action is to wait", action is PingUploadTask.Wait)
val waitTime = (action as PingUploadTask.Wait).time
assertTrue("Waiting for more than 50s, was: $waitTime", waitTime > 50_000)
assertTrue("Waiting for less than a minute, was: $waitTime", waitTime <= 60_000)
}
}

40
glean-core/android/src/test/java/mozilla/telemetry/glean/GleanTest.kt
Просмотреть файл

@ -14,6 +14,7 @@ import kotlinx.coroutines.Dispatchers as KotlinDispatchers
import kotlinx.coroutines.launch
import kotlinx.coroutines.ObsoleteCoroutinesApi
import kotlinx.coroutines.runBlocking
import mozilla.telemetry.glean.internal.gleanSetTestMode
import mozilla.telemetry.glean.GleanMetrics.GleanError
import mozilla.telemetry.glean.GleanMetrics.GleanInternalMetrics
import mozilla.telemetry.glean.GleanMetrics.Pings
@ -68,7 +69,6 @@ class GleanTest {
Glean.setUploadEnabled(true)
}
// New from glean-core.
@Test
fun `send a ping`() {
delayMetricsPing(context)
@ -81,9 +81,10 @@ class GleanTest {
// Trigger it to upload
triggerWorkManager(context)
// Make sure the number of request received thus far is only 1,
// we are only expecting one baseline ping.
assertEquals(server.requestCount, 2)
// We got exactly 2 pings here:
// 1. The overdue metrics ping from the initial `Glean` setup before we reset it.
// 2. The baseline ping triggered by the background event above.
assertEquals(2, server.requestCount)
var request = server.takeRequest(20L, TimeUnit.SECONDS)!!
var docType = request.path!!.split("/")[3]
@ -115,6 +116,7 @@ class GleanTest {
// Tests from glean-ac (706af1f).
@Test
@Ignore("needs UniFFI migration")
fun `disabling upload should disable metrics recording`() {
val stringMetric = StringMetricType(
disabled = false,
@ -156,7 +158,8 @@ class GleanTest {
fun `test experiments recording before Glean inits`() {
// This test relies on Glean not being initialized and task queuing to be on.
Glean.testDestroyGleanHandle()
Dispatchers.API.setTaskQueueing(true)
// We're reaching into internals, but `resetGlean` will re-enable test mode later.
gleanSetTestMode(false)
Glean.setExperimentActive(
"experiment_set_preinit", "branch_a"
@ -177,6 +180,7 @@ class GleanTest {
// Suppressing our own deprecation before we move over to the new event recording API.
@Test
@Ignore("needs UniFFI migration")
@Suppress("ComplexMethod", "LongMethod", "NestedBlockDepth", "DEPRECATION")
fun `test sending of foreground and background pings`() {
val server = getMockWebServer()
@ -262,8 +266,9 @@ class GleanTest {
@Test
fun `test sending of startup baseline ping`() {
// TODO: Should be in Rust now.
// Set the dirty flag.
LibGleanFFI.INSTANCE.glean_set_dirty_flag(true.toByte())
Glean.handleForegroundEvent()
// Restart glean and don't clear the stores.
val server = getMockWebServer()
@ -314,6 +319,7 @@ class GleanTest {
}
@Test
@Ignore("needs UniFFI migration")
fun `queued recorded metrics correctly record during init`() {
val counterMetric = CounterMetricType(
disabled = false,
@ -360,6 +366,7 @@ class GleanTest {
}
@Test
@Ignore("needs UniFFI migration")
fun `Don't handle events when uninitialized`() {
val gleanSpy = spy<GleanInternalAPI>(GleanInternalAPI::class.java)
@ -370,6 +377,7 @@ class GleanTest {
}
@Test
@Ignore("needs UniFFI migration")
fun `The appChannel must be correctly set, if requested`() {
// No appChannel must be set if nothing was provided through the config
// options.
@ -393,7 +401,7 @@ class GleanTest {
// 1539480 BACKWARD COMPATIBILITY HACK that is not needed anymore.
@Test
fun `getLanguageTag() reports the tag for the default locale`() {
fun `getLanguageTag reports the tag for the default locale`() {
val defaultLanguageTag = getLocaleTag()
assertNotNull(defaultLanguageTag)
@ -419,6 +427,7 @@ class GleanTest {
}
@Test
@Ignore("needs UniFFI migration")
fun `getLanguage reports the modern translation for some languages`() {
assertEquals("he", getLanguageFromLocale(Locale("iw", "IL")))
assertEquals("id", getLanguageFromLocale(Locale("in", "ID")))
@ -426,6 +435,7 @@ class GleanTest {
}
@Test
@Ignore("needs UniFFI migration")
fun `ping collection must happen after currently scheduled metrics recordings`() {
// Given the following block of code:
//
@ -491,6 +501,7 @@ class GleanTest {
}
@Test
@Ignore("needs UniFFI migration")
fun `Basic metrics should be cleared when disabling uploading`() {
val stringMetric = StringMetricType(
disabled = false,
@ -515,6 +526,7 @@ class GleanTest {
}
@Test
@Ignore("needs UniFFI migration")
fun `Core metrics should be cleared and restored when disabling and enabling uploading`() {
assertTrue(GleanInternalMetrics.os.testHasValue())
@ -528,7 +540,7 @@ class GleanTest {
@Test
fun `Workers should be cancelled when disabling uploading`() {
// Force the MetricsPingScheduler to schedule the MetricsPingWorker
Glean.metricsPingScheduler.schedulePingCollection(
Glean.metricsPingScheduler!!.schedulePingCollection(
Calendar.getInstance(),
true,
Pings.metricsReasonCodes.overdue
@ -540,7 +552,7 @@ class GleanTest {
assertTrue("PingUploadWorker is enqueued",
getWorkerStatus(context, PingUploadWorker.PING_WORKER_TAG).isEnqueued)
assertTrue("MetricsPingWorker is enqueued",
Glean.metricsPingScheduler.timer != null)
Glean.metricsPingScheduler!!.timer != null)
Glean.setUploadEnabled(true)
@ -550,14 +562,14 @@ class GleanTest {
assertTrue("PingUploadWorker is enqueued",
getWorkerStatus(context, PingUploadWorker.PING_WORKER_TAG).isEnqueued)
assertTrue("MetricsPingWorker is enqueued",
Glean.metricsPingScheduler.timer != null)
Glean.metricsPingScheduler!!.timer != null)
// Toggle upload enabled to false
Glean.setUploadEnabled(false)
// Verify workers have been cancelled
assertTrue("MetricsPingWorker is not enqueued",
Glean.metricsPingScheduler.timer == null)
Glean.metricsPingScheduler!!.timer == null)
}
@Test
@ -584,6 +596,7 @@ class GleanTest {
}
@Test
@Ignore("needs UniFFI migration")
fun `overflowing the task queue records telemetry`() {
delayMetricsPing(context)
val server = getMockWebServer()
@ -628,6 +641,7 @@ class GleanTest {
}
@Test
@Ignore("needs UniFFI migration")
fun `sending deletion ping if disabled outside of run`() {
val server = getMockWebServer()
resetGlean(
@ -679,6 +693,7 @@ class GleanTest {
}
@Test
@Ignore("needs UniFFI migration")
fun `test sending of startup baseline ping with application lifetime metric`() {
// Set the dirty flag.
LibGleanFFI.INSTANCE.glean_set_dirty_flag(true.toByte())
@ -721,6 +736,7 @@ class GleanTest {
}
@Test
@Ignore("needs UniFFI migration")
fun `test dirty flag is reset to false`() {
// Set the dirty flag.
LibGleanFFI.INSTANCE.glean_set_dirty_flag(true.toByte())
@ -731,6 +747,7 @@ class GleanTest {
}
@Test
@Ignore("needs UniFFI migration")
fun `setting debugViewTag before initialization should not crash`() {
// Can't use resetGlean directly
Glean.testDestroyGleanHandle()
@ -826,6 +843,7 @@ class GleanTest {
}
@Test
@Ignore("needs UniFFI migration")
fun `test passing in explicit BuildInfo`() {
Glean.testDestroyGleanHandle()
Glean.initialize(

1
glean-core/android/src/test/java/mozilla/telemetry/glean/UniffiTest.kt
Просмотреть файл

@ -21,7 +21,6 @@ import mozilla.telemetry.glean.config.Configuration
import mozilla.telemetry.glean.internal.CounterMetric
import mozilla.telemetry.glean.internal.Lifetime
import mozilla.telemetry.glean.internal.CommonMetricData
import mozilla.telemetry.glean.rust.LibGleanFFI
import mozilla.telemetry.glean.rust.toBoolean
import mozilla.telemetry.glean.rust.toByte
import mozilla.telemetry.glean.scheduler.GleanLifecycleObserver

8
glean-core/android/src/test/java/mozilla/telemetry/glean/debug/GleanDebugActivityTest.kt
Просмотреть файл

@ -29,7 +29,7 @@ import mozilla.telemetry.glean.getMockWebServer
import mozilla.telemetry.glean.net.HeadersList
import mozilla.telemetry.glean.net.PingUploader
import mozilla.telemetry.glean.net.UploadResult
import mozilla.telemetry.glean.net.HttpResponse
import mozilla.telemetry.glean.net.HttpStatus
import mozilla.telemetry.glean.private.NoReasonCodes
import mozilla.telemetry.glean.private.PingType
import mozilla.telemetry.glean.testing.GleanTestRule
@ -50,14 +50,14 @@ private class TestPingTagClient(
url.startsWith(responseUrl))
debugHeaderValue?.let {
assertEquals("The debug view header must match what the ping tag was set to",
debugHeaderValue, headers.find { it.first == "X-Debug-ID" }!!.second)
debugHeaderValue, headers.get("X-Debug-ID")!!)
}
sourceTagsValue?.let {
assertEquals("The source tags header must match what the ping tag was set to",
sourceTagsValue.joinToString(","), headers.find { it.first == "X-Source-Tags" }!!.second)
sourceTagsValue.joinToString(","), headers.get("X-Source-Tags")!!)
}
return HttpResponse(200)
return HttpStatus(200)
}
}

4
glean-core/android/src/test/java/mozilla/telemetry/glean/net/BaseUploaderTest.kt
Просмотреть файл

@ -17,7 +17,7 @@ import org.robolectric.RobolectricTestRunner
class BaseUploaderTest {
private val testPath: String = "/some/random/path/not/important"
private val testPing: String = "{ 'ping': 'test' }"
private val testHeaders: HeadersList = mutableListOf(Pair("X-Test-Glean", "nothing-to-see-here"))
private val testHeaders: HeadersList = mutableMapOf("X-Test-Glean" to "nothing-to-see-here")
private val testDefaultConfig = Configuration()
/**
@ -25,7 +25,7 @@ class BaseUploaderTest {
*/
private class TestUploader : PingUploader {
override fun upload(url: String, data: ByteArray, headers: HeadersList): UploadResult {
return UnrecoverableFailure
return UnrecoverableFailure(0)
}
}

14
glean-core/android/src/test/java/mozilla/telemetry/glean/net/HttpURLConnectionUploaderTest.kt
Просмотреть файл

@ -46,7 +46,7 @@ class HttpURLConnectionUploaderTest {
doReturn(connection).`when`(client).openConnection(anyString())
doReturn(200).`when`(client).doUpload(connection, testPing.toByteArray(Charsets.UTF_8))
client.upload(testPath, testPing.toByteArray(Charsets.UTF_8), emptyList())
client.upload(testPath, testPing.toByteArray(Charsets.UTF_8), emptyMap())
verify<HttpURLConnection>(connection).readTimeout = HttpURLConnectionUploader.DEFAULT_READ_TIMEOUT
verify<HttpURLConnection>(connection).connectTimeout = HttpURLConnectionUploader.DEFAULT_CONNECTION_TIMEOUT
@ -65,7 +65,7 @@ class HttpURLConnectionUploaderTest {
"Test-header" to "SomeValue",
"OtherHeader" to "Glean/Test 25.0.2"
)
uploader.upload(testPath, testPing.toByteArray(Charsets.UTF_8), expectedHeaders.toList())
uploader.upload(testPath, testPing.toByteArray(Charsets.UTF_8), expectedHeaders)
val headerNameCaptor = argumentCaptor<String>()
val headerValueCaptor = argumentCaptor<String>()
@ -100,8 +100,8 @@ class HttpURLConnectionUploaderTest {
doReturn(connection).`when`(client).openConnection(anyString())
assertEquals(
client.upload(testPath, testPing.toByteArray(Charsets.UTF_8), emptyList()),
HttpResponse(200)
client.upload(testPath, testPing.toByteArray(Charsets.UTF_8), emptyMap()),
HttpStatus(200)
)
verify<HttpURLConnection>(connection, times(1)).disconnect()
}
@ -117,7 +117,7 @@ class HttpURLConnectionUploaderTest {
val client = HttpURLConnectionUploader()
val url = testConfig.serverEndpoint + testPath
assertNotNull(client.upload(url, testPing.toByteArray(Charsets.UTF_8), emptyList()))
assertNotNull(client.upload(url, testPing.toByteArray(Charsets.UTF_8), emptyMap()))
val request = server.takeRequest()
assertEquals(testPath, request.path)
@ -172,7 +172,7 @@ class HttpURLConnectionUploaderTest {
// Trigger the connection.
val url = testConfig.serverEndpoint + testPath
val client = HttpURLConnectionUploader()
assertNotNull(client.upload(url, testPing.toByteArray(Charsets.UTF_8), emptyList()))
assertNotNull(client.upload(url, testPing.toByteArray(Charsets.UTF_8), emptyMap()))
val request = server.takeRequest()
assertEquals(testPath, request.path)
@ -191,6 +191,6 @@ class HttpURLConnectionUploaderTest {
fun `upload() discards pings on malformed URLs`() {
val client = spy<HttpURLConnectionUploader>(HttpURLConnectionUploader())
doThrow(MalformedURLException()).`when`(client).openConnection(anyString())
assertEquals(UnrecoverableFailure, client.upload("path", "ping".toByteArray(Charsets.UTF_8), emptyList()))
assertEquals(UnrecoverableFailure(0), client.upload("path", "ping".toByteArray(Charsets.UTF_8), emptyMap()))
}
}

16
glean-core/android/src/test/java/mozilla/telemetry/glean/scheduler/MetricsPingSchedulerTest.kt
Просмотреть файл

@ -289,7 +289,7 @@ class MetricsPingSchedulerTest {
assertTrue("The initial test data must have been recorded", testMetric.testHasValue())
// Manually call the function to trigger the collection.
Glean.metricsPingScheduler.collectPingAndReschedule(
Glean.metricsPingScheduler!!.collectPingAndReschedule(
Calendar.getInstance(),
false,
Pings.metricsReasonCodes.overdue
@ -634,19 +634,19 @@ class MetricsPingSchedulerTest {
Glean.metricsPingScheduler = MetricsPingScheduler(context, GleanBuildInfo.buildInfo)
// No work should be enqueued at the beginning of the test.
assertNull(Glean.metricsPingScheduler.timer)
assertNull(Glean.metricsPingScheduler!!.timer)
// Manually schedule a collection task for today.
Glean.metricsPingScheduler.schedulePingCollection(
Glean.metricsPingScheduler!!.schedulePingCollection(
Calendar.getInstance(),
sendTheNextCalendarDay = false,
reason = Pings.metricsReasonCodes.overdue
)
// We expect the worker to be scheduled.
assertNotNull(Glean.metricsPingScheduler.timer)
assertNotNull(Glean.metricsPingScheduler!!.timer)
Glean.metricsPingScheduler.cancel()
Glean.metricsPingScheduler!!.cancel()
resetGlean(clearStores = true)
}
@ -660,14 +660,14 @@ class MetricsPingSchedulerTest {
// Verify that the worker is enqueued
assertNotNull("MetricsPingWorker is enqueued",
Glean.metricsPingScheduler.timer)
Glean.metricsPingScheduler!!.timer)
// Cancel the worker
Glean.metricsPingScheduler.cancel()
Glean.metricsPingScheduler!!.cancel()
// Verify worker has been cancelled
assertNull("MetricsPingWorker is not enqueued",
Glean.metricsPingScheduler.timer)
Glean.metricsPingScheduler!!.timer)
}
@Test

29
glean-core/src/glean.udl
Просмотреть файл

@ -22,10 +22,13 @@ namespace glean {
void glean_handle_client_inactive();
void glean_submit_ping_by_name(string ping_name, optional string? reason = null);
void glean_submit_ping_by_name_sync(string ping_name, optional string? reason = null);
boolean glean_submit_ping_by_name_sync(string ping_name, optional string? reason = null);
void glean_set_test_mode(boolean enabled);
void glean_test_destroy_glean(boolean clear_stores);
PingUploadTask glean_get_upload_task();
void glean_process_ping_upload_response(string uuid, UploadResult result);
};
// The Glean configuration.
@ -82,6 +85,9 @@ callback interface OnGleanEvents {
//
// The language SDK
void start_metrics_ping_scheduler();
// Called when upload is disabled and uploads should be stopped
void cancel_uploads();
};
dictionary RecordedExperiment {
@ -89,6 +95,27 @@ dictionary RecordedExperiment {
record<DOMString, string>? extra;
};
dictionary PingRequest {
string document_id;
string path;
sequence<u8> body;
record<DOMString, string> headers;
};
[Enum]
interface PingUploadTask {
Upload(PingRequest request);
Wait(u64 time);
Done(i8 unused);
};
[Enum]
interface UploadResult {
RecoverableFailure(i8 unused);
UnrecoverableFailure(i8 unused);
HttpStatus(i32 code);
};
enum Lifetime {
"Ping",
"Application",

81
glean-core/src/lib.rs
Просмотреть файл

@ -49,6 +49,7 @@ use crate::core_metrics::ClientInfoMetrics;
pub use crate::error::{Error, ErrorKind, Result};
pub use crate::error_recording::{test_get_num_recorded_errors, ErrorType};
pub use crate::metrics::{CounterMetric, PingType, RecordedExperiment};
pub use crate::upload::{PingRequest, PingUploadTask, UploadResult};
const GLEAN_VERSION: &str = env!("CARGO_PKG_VERSION");
const GLEAN_SCHEMA_VERSION: u32 = 1;
@ -187,6 +188,9 @@ pub trait OnGleanEvents: Send {
/// Start the Metrics Ping Scheduler.
fn start_metrics_ping_scheduler(&self);
/// Called when upload is disabled and uploads should be stopped
fn cancel_uploads(&self);
}
/// Initializes Glean.
@ -212,10 +216,10 @@ fn initialize_inner(
) -> Result<()> {
if was_initialize_called() {
log::error!("Glean should not be initialized multiple times");
todo!();
return Ok(());
}
let _init_handle = std::thread::Builder::new()
let init_handle = std::thread::Builder::new()
.name("glean.init".into())
.spawn(move || {
let upload_enabled = cfg.upload_enabled;
@ -238,6 +242,8 @@ fn initialize_inner(
callbacks,
});
let mut is_first_run = false;
let mut dirty_flag = false;
core::with_glean_mut(|glean| {
let state = global_state().lock().unwrap();
@ -271,7 +277,7 @@ fn initialize_inner(
// `false` so that dirty startup pings won't be sent if Glean
// initialization does not complete successfully.
// TODO Bug 1672956 will decide where to set this flag again.
let dirty_flag = glean.is_dirty_flag_set();
dirty_flag = glean.is_dirty_flag_set();
glean.set_dirty_flag(false);
// Perform registration of pings that were attempted to be
@ -288,7 +294,7 @@ fn initialize_inner(
// If this is the first time ever the Glean SDK runs, make sure to set
// some initial core metrics in case we need to generate early pings.
// The next times we start, we would have them around already.
let is_first_run = glean.is_first_run();
is_first_run = glean.is_first_run();
if is_first_run {
initialize_core_metrics(glean, &state.client_info);
}
@ -302,11 +308,27 @@ fn initialize_inner(
if pings_submitted || !upload_enabled {
state.callbacks.trigger_upload();
}
});
// The metrics ping scheduler might _synchronously_ submit a ping
// so that it runs before we clear application-lifetime metrics further below.
// For that it needs access to the `Glean` object.
// Thus we need to unlock that by leaving the context above,
// then re-lock it afterwards.
// That's safe because user-visible functions will be queued and thus not execute until
// we unblock later anyway.
{
let state = global_state().lock().unwrap();
// Set up information and scheduling for Glean owned pings. Ideally, the "metrics"
// ping startup check should be performed before any other ping, since it relies
// on being dispatched to the API context before any other metric.
state.callbacks.start_metrics_ping_scheduler();
state.callbacks.trigger_upload();
}
core::with_glean_mut(|glean| {
let state = global_state().lock().unwrap();
// Check if the "dirty flag" is set. That means the product was probably
// force-closed. If that's the case, submit a 'baseline' ping with the
@ -348,6 +370,13 @@ fn initialize_inner(
})
.expect("Failed to spawn Glean's init thread");
// In test mode we wait for initialization to finish.
if dispatcher::global::is_test_mode() {
init_handle
.join()
.expect("Failed to join initialization thread");
}
// Mark the initialization as called: this needs to happen outside of the
// dispatched block!
INITIALIZE_CALLED.store(true, Ordering::SeqCst);
@ -403,20 +432,26 @@ pub fn glean_enable_logging() {
/// Sets whether upload is enabled or not.
pub fn glean_set_upload_enabled(enabled: bool) {
core::with_glean_mut(|glean| {
if !was_initialize_called() {
return;
}
crate::launch_with_glean_mut(move |glean| {
let state = global_state().lock().unwrap();
let original_enabled = glean.is_upload_enabled();
if !enabled {
//changes_callback.will_be_disabled();
state.callbacks.cancel_uploads();
}
glean.set_upload_enabled(enabled);
if !original_enabled && enabled {
//changes_callback.on_enabled();
initialize_core_metrics(glean, &state.client_info);
}
if original_enabled && !enabled {
//changes_callback.on_disabled();
state.callbacks.trigger_upload();
}
})
}
@ -574,16 +609,7 @@ pub fn glean_handle_client_inactive() {
/// Collect and submit a ping for eventual upload by name.
pub fn glean_submit_ping_by_name(ping_name: String, reason: Option<String>) {
dispatcher::launch(|| glean_submit_ping_by_name_sync(ping_name, reason))
}
/// Collect and submit a ping (by its name) for eventual upload, synchronously.
pub fn glean_submit_ping_by_name_sync(ping_name: String, reason: Option<String>) {
core::with_glean(|glean| {
if !glean.is_upload_enabled() {
return;
}
crate::launch_with_glean(move |glean| {
let sent = glean.submit_ping_by_name(&ping_name, reason.as_deref());
if sent {
let state = global_state().lock().unwrap();
@ -592,6 +618,15 @@ pub fn glean_submit_ping_by_name_sync(ping_name: String, reason: Option<String>)
})
}
/// Collect and submit a ping (by its name) for eventual upload, synchronously.
///
/// Note: This does not trigger the uploader. The caller is responsible to do this.
pub fn glean_submit_ping_by_name_sync(ping_name: String, reason: Option<String>) -> bool {
core::with_glean(|glean| {
glean.submit_ping_by_name(&ping_name, reason.as_deref())
})
}
/// **TEST-ONLY Method**
///
/// Enable test mode
@ -622,6 +657,16 @@ pub fn glean_test_destroy_glean(clear_stores: bool) {
INITIALIZE_CALLED.store(false, Ordering::SeqCst);
}
/// Get the next upload task
pub fn glean_get_upload_task() -> PingUploadTask {
core::with_glean(|glean| glean.get_upload_task())
}
/// Processes the response from an attempt to upload a ping.
pub fn glean_process_ping_upload_response(uuid: String, result: UploadResult) {
core::with_glean(|glean| glean.process_ping_upload_response(&uuid, result))
}
#[allow(missing_docs)]
mod ffi {
use super::*;

152
glean-core/src/upload/mod.rs
Просмотреть файл

@ -129,15 +129,21 @@ impl RateLimiter {
/// If new variants are added, this should be reflected in `glean-core/ffi/src/upload.rs` as well.
#[derive(PartialEq, Debug)]
pub enum PingUploadTask {
/// A PingRequest popped from the front of the queue.
/// See [`PingRequest`](struct.PingRequest.html) for more information.
Upload(PingRequest),
/// An upload task
Upload {
/// The ping request for upload
/// See [`PingRequest`](struct.PingRequest.html) for more information.
request: PingRequest,
},
/// A flag signaling that the pending pings directories are not done being processed,
/// thus the requester should wait and come back later.
///
/// Contains the amount of time in milliseconds
/// the requester should wait before requesting a new task.
Wait(u64),
Wait {
/// The time in milliseconds
/// the requester should wait before requesting a new task.
time: u64,
},
/// A flag signaling that requester doesn't need to request any more upload tasks at this moment.
///
/// There are three possibilities for this scenario:
@ -150,18 +156,26 @@ pub enum PingUploadTask {
/// An "uploading window" starts when a requester gets a new
/// `PingUploadTask::Upload(PingRequest)` response and finishes when they
/// finally get a `PingUploadTask::Done` or `PingUploadTask::Wait` response.
Done,
Done {
#[doc(hidden)]
/// Unused field. Required because UniFFI can't handle variants without fields.
unused: i8,
},
}
impl PingUploadTask {
/// Whether the current task is an upload task.
pub fn is_upload(&self) -> bool {
matches!(self, PingUploadTask::Upload(_))
matches!(self, PingUploadTask::Upload { .. })
}
/// Whether the current task is wait task.
pub fn is_wait(&self) -> bool {
matches!(self, PingUploadTask::Wait(_))
matches!(self, PingUploadTask::Wait { .. })
}
fn done() -> Self {
PingUploadTask::Done { unused: 0 }
}
}
@ -508,9 +522,9 @@ impl PingUploadManager {
let wait_or_done = |time: u64| {
self.wait_attempt_count.fetch_add(1, Ordering::SeqCst);
if self.wait_attempt_count() > self.policy.max_wait_attempts() {
PingUploadTask::Done
PingUploadTask::done()
} else {
PingUploadTask::Wait(time)
PingUploadTask::Wait { time }
}
};
@ -528,7 +542,7 @@ impl PingUploadManager {
log::warn!(
"Reached maximum recoverable failures for the current uploading window. You are done."
);
return PingUploadTask::Done;
return PingUploadTask::done();
}
let mut queue = self
@ -563,11 +577,13 @@ impl PingUploadManager {
}
}
PingUploadTask::Upload(queue.pop_front().unwrap())
PingUploadTask::Upload {
request: queue.pop_front().unwrap(),
}
}
None => {
log::info!("No more pings to upload! You are done.");
PingUploadTask::Done
PingUploadTask::done()
}
}
}
@ -648,12 +664,12 @@ impl PingUploadManager {
}
match status {
HttpStatus(status @ 200..=299) => {
log::info!("Ping {} successfully sent {}.", document_id, status);
HttpStatus { code: 200..=299 } => {
log::info!("Ping {} successfully sent.", document_id);
self.directory_manager.delete_file(document_id);
}
UnrecoverableFailure | HttpStatus(400..=499) => {
UnrecoverableFailure { .. } | HttpStatus { code: 400..=499 } => {
log::warn!(
"Unrecoverable upload failure while attempting to send ping {}. Error was {:?}",
document_id,
@ -662,7 +678,7 @@ impl PingUploadManager {
self.directory_manager.delete_file(document_id);
}
RecoverableFailure | HttpStatus(_) => {
RecoverableFailure { .. } | HttpStatus { .. } => {
log::warn!(
"Recoverable upload failure while attempting to send ping {}, will retry. Error was {:?}",
document_id,
@ -749,7 +765,6 @@ mod test {
use uuid::Uuid;
use super::UploadResult::*;
use super::*;
use crate::metrics::PingType;
use crate::{tests::new_glean, PENDING_PINGS_DIRECTORY};
@ -762,7 +777,7 @@ mod test {
// Try and get the next request.
// Verify request was not returned
assert_eq!(glean.get_upload_task(), PingUploadTask::Done);
assert_eq!(glean.get_upload_task(), PingUploadTask::done());
}
#[test]
@ -801,7 +816,7 @@ mod test {
// Verify that after all requests are returned, none are left
assert_eq!(
upload_manager.get_upload_task(&glean, false),
PingUploadTask::Done
PingUploadTask::done()
);
}
@ -831,7 +846,7 @@ mod test {
// Verify that we are indeed told to wait because we are at capacity
match upload_manager.get_upload_task(&glean, false) {
PingUploadTask::Wait(time) => {
PingUploadTask::Wait { time } => {
// Wait for the uploading window to reset
thread::sleep(Duration::from_millis(time));
}
@ -859,7 +874,7 @@ mod test {
// Verify there really isn't any ping in the queue
assert_eq!(
upload_manager.get_upload_task(&glean, false),
PingUploadTask::Done
PingUploadTask::done()
);
}
@ -887,12 +902,12 @@ mod test {
let upload_task = glean.get_upload_task();
match upload_task {
PingUploadTask::Upload(request) => assert!(request.is_deletion_request()),
PingUploadTask::Upload { request } => assert!(request.is_deletion_request()),
_ => panic!("Expected upload manager to return the next request!"),
}
// Verify there really isn't any other pings in the queue
assert_eq!(glean.get_upload_task(), PingUploadTask::Done);
assert_eq!(glean.get_upload_task(), PingUploadTask::done());
}
#[test]
@ -921,7 +936,7 @@ mod test {
// Verify that after all requests are returned, none are left
assert_eq!(
upload_manager.get_upload_task(&glean, false),
PingUploadTask::Done
PingUploadTask::done()
);
}
@ -941,10 +956,10 @@ mod test {
// Get the submitted PingRequest
match glean.get_upload_task() {
PingUploadTask::Upload(request) => {
PingUploadTask::Upload { request } => {
// Simulate the processing of a sucessfull request
let document_id = request.document_id;
glean.process_ping_upload_response(&document_id, HttpStatus(200));
glean.process_ping_upload_response(&document_id, UploadResult::http_status(200));
// Verify file was deleted
assert!(!pending_pings_dir.join(document_id).exists());
}
@ -952,7 +967,7 @@ mod test {
}
// Verify that after request is returned, none are left
assert_eq!(glean.get_upload_task(), PingUploadTask::Done);
assert_eq!(glean.get_upload_task(), PingUploadTask::done());
}
#[test]
@ -971,10 +986,10 @@ mod test {
// Get the submitted PingRequest
match glean.get_upload_task() {
PingUploadTask::Upload(request) => {
PingUploadTask::Upload { request } => {
// Simulate the processing of a client error
let document_id = request.document_id;
glean.process_ping_upload_response(&document_id, HttpStatus(404));
glean.process_ping_upload_response(&document_id, UploadResult::http_status(404));
// Verify file was deleted
assert!(!pending_pings_dir.join(document_id).exists());
}
@ -982,7 +997,7 @@ mod test {
}
// Verify that after request is returned, none are left
assert_eq!(glean.get_upload_task(), PingUploadTask::Done);
assert_eq!(glean.get_upload_task(), PingUploadTask::done());
}
#[test]
@ -998,13 +1013,13 @@ mod test {
// Get the submitted PingRequest
match glean.get_upload_task() {
PingUploadTask::Upload(request) => {
PingUploadTask::Upload { request } => {
// Simulate the processing of a client error
let document_id = request.document_id;
glean.process_ping_upload_response(&document_id, HttpStatus(500));
glean.process_ping_upload_response(&document_id, UploadResult::http_status(500));
// Verify this ping was indeed re-enqueued
match glean.get_upload_task() {
PingUploadTask::Upload(request) => {
PingUploadTask::Upload { request } => {
assert_eq!(document_id, request.document_id);
}
_ => panic!("Expected upload manager to return the next request!"),
@ -1014,7 +1029,7 @@ mod test {
}
// Verify that after request is returned, none are left
assert_eq!(glean.get_upload_task(), PingUploadTask::Done);
assert_eq!(glean.get_upload_task(), PingUploadTask::done());
}
#[test]
@ -1033,10 +1048,13 @@ mod test {
// Get the submitted PingRequest
match glean.get_upload_task() {
PingUploadTask::Upload(request) => {
PingUploadTask::Upload { request } => {
// Simulate the processing of a client error
let document_id = request.document_id;
glean.process_ping_upload_response(&document_id, UnrecoverableFailure);
glean.process_ping_upload_response(
&document_id,
UploadResult::unrecoverable_failure(),
);
// Verify file was deleted
assert!(!pending_pings_dir.join(document_id).exists());
}
@ -1044,7 +1062,7 @@ mod test {
}
// Verify that after request is returned, none are left
assert_eq!(glean.get_upload_task(), PingUploadTask::Done);
assert_eq!(glean.get_upload_task(), PingUploadTask::done());
}
#[test]
@ -1064,7 +1082,7 @@ mod test {
// Try and get the first request.
let req = match upload_manager.get_upload_task(&glean, false) {
PingUploadTask::Upload(req) => req,
PingUploadTask::Upload { request } => request,
_ => panic!("Expected upload manager to return the next request!"),
};
assert_eq!(doc1, req.document_id);
@ -1073,22 +1091,30 @@ mod test {
upload_manager.enqueue_ping(&glean, &doc2, &path2, "", None);
// Mark as processed
upload_manager.process_ping_upload_response(&glean, &req.document_id, HttpStatus(200));
upload_manager.process_ping_upload_response(
&glean,
&req.document_id,
UploadResult::http_status(200),
);
// Get the second request.
let req = match upload_manager.get_upload_task(&glean, false) {
PingUploadTask::Upload(req) => req,
PingUploadTask::Upload { request } => request,
_ => panic!("Expected upload manager to return the next request!"),
};
assert_eq!(doc2, req.document_id);
// Mark as processed
upload_manager.process_ping_upload_response(&glean, &req.document_id, HttpStatus(200));
upload_manager.process_ping_upload_response(
&glean,
&req.document_id,
UploadResult::http_status(200),
);
// ... and then we're done.
assert_eq!(
upload_manager.get_upload_task(&glean, false),
PingUploadTask::Done
PingUploadTask::done()
);
}
@ -1107,7 +1133,7 @@ mod test {
// Get the submitted PingRequest
match glean.get_upload_task() {
PingUploadTask::Upload(request) => {
PingUploadTask::Upload { request } => {
assert_eq!(request.headers.get("X-Debug-ID").unwrap(), "valid-tag")
}
_ => panic!("Expected upload manager to return the next request!"),
@ -1136,7 +1162,7 @@ mod test {
// There should be no more queued tasks
assert_eq!(
upload_manager.get_upload_task(&glean, false),
PingUploadTask::Done
PingUploadTask::done()
);
}
@ -1165,10 +1191,10 @@ mod test {
// Return the max recoverable error failures in a row
for _ in 0..max_recoverable_failures {
match upload_manager.get_upload_task(&glean, false) {
PingUploadTask::Upload(req) => upload_manager.process_ping_upload_response(
PingUploadTask::Upload { request } => upload_manager.process_ping_upload_response(
&glean,
&req.document_id,
RecoverableFailure,
&request.document_id,
UploadResult::recoverable_failure(),
),
_ => panic!("Expected upload manager to return the next request!"),
}
@ -1178,7 +1204,7 @@ mod test {
// we are done even though we haven't gotten all the enqueued requests.
assert_eq!(
upload_manager.get_upload_task(&glean, false),
PingUploadTask::Done
PingUploadTask::done()
);
// Verify all requests are returned when we try again.
@ -1226,7 +1252,7 @@ mod test {
// One ping should have been enqueued.
// Make sure it is the newest ping.
match upload_manager.get_upload_task(&glean, false) {
PingUploadTask::Upload(request) => assert_eq!(&request.document_id, newest_ping_id),
PingUploadTask::Upload { request } => assert_eq!(&request.document_id, newest_ping_id),
_ => panic!("Expected upload manager to return the next request!"),
}
@ -1234,7 +1260,7 @@ mod test {
// they should all have been deleted because pending pings quota was hit.
assert_eq!(
upload_manager.get_upload_task(&glean, false),
PingUploadTask::Done
PingUploadTask::done()
);
// Verify that the correct number of deleted pings was recorded
@ -1297,7 +1323,7 @@ mod test {
// Make sure it is the newest ping.
for ping_id in expected_pings.iter().rev() {
match upload_manager.get_upload_task(&glean, false) {
PingUploadTask::Upload(request) => assert_eq!(&request.document_id, ping_id),
PingUploadTask::Upload { request } => assert_eq!(&request.document_id, ping_id),
_ => panic!("Expected upload manager to return the next request!"),
}
}
@ -1306,7 +1332,7 @@ mod test {
// they should all have been deleted because pending pings quota was hit.
assert_eq!(
upload_manager.get_upload_task(&glean, false),
PingUploadTask::Done
PingUploadTask::done()
);
// Verify that the correct number of deleted pings was recorded
@ -1371,7 +1397,7 @@ mod test {
// Make sure it is the newest ping.
for ping_id in expected_pings.iter().rev() {
match upload_manager.get_upload_task(&glean, false) {
PingUploadTask::Upload(request) => assert_eq!(&request.document_id, ping_id),
PingUploadTask::Upload { request } => assert_eq!(&request.document_id, ping_id),
_ => panic!("Expected upload manager to return the next request!"),
}
}
@ -1380,7 +1406,7 @@ mod test {
// they should all have been deleted because pending pings quota was hit.
assert_eq!(
upload_manager.get_upload_task(&glean, false),
PingUploadTask::Done
PingUploadTask::done()
);
// Verify that the correct number of deleted pings was recorded
@ -1445,7 +1471,7 @@ mod test {
// Make sure it is the newest ping.
for ping_id in expected_pings.iter().rev() {
match upload_manager.get_upload_task(&glean, false) {
PingUploadTask::Upload(request) => assert_eq!(&request.document_id, ping_id),
PingUploadTask::Upload { request } => assert_eq!(&request.document_id, ping_id),
_ => panic!("Expected upload manager to return the next request!"),
}
}
@ -1454,7 +1480,7 @@ mod test {
// they should all have been deleted because pending pings quota was hit.
assert_eq!(
upload_manager.get_upload_task(&glean, false),
PingUploadTask::Done
PingUploadTask::done()
);
// Verify that the correct number of deleted pings was recorded
@ -1503,7 +1529,7 @@ mod test {
// Get the first ping, it should be returned normally.
match upload_manager.get_upload_task(&glean, false) {
PingUploadTask::Upload(_) => {}
PingUploadTask::Upload { .. } => {}
_ => panic!("Expected upload manager to return the next request!"),
}
@ -1519,7 +1545,7 @@ mod test {
// we then get PingUploadTask::Done.
assert_eq!(
upload_manager.get_upload_task(&glean, false),
PingUploadTask::Done
PingUploadTask::done()
);
// Wait for the rate limiter to allow upload tasks again.
@ -1532,7 +1558,7 @@ mod test {
// And once we are done we don't need to wait anymore.
assert_eq!(
upload_manager.get_upload_task(&glean, false),
PingUploadTask::Done
PingUploadTask::done()
);
}
@ -1541,7 +1567,7 @@ mod test {
let (glean, dir) = new_glean(None);
let upload_manager = PingUploadManager::new(dir.path(), "test");
match upload_manager.get_upload_task(&glean, false) {
PingUploadTask::Wait(time) => {
PingUploadTask::Wait { time } => {
assert_eq!(time, WAIT_TIME_FOR_PING_PROCESSING);
}
_ => panic!("Expected upload manager to return a wait task!"),

49
glean-core/src/upload/result.rs
Просмотреть файл

@ -35,17 +35,28 @@ pub enum UploadResult {
/// During upload something went wrong,
/// e.g. the network connection failed.
/// The upload should be retried at a later time.
RecoverableFailure,
RecoverableFailure {
#[doc(hidden)]
/// Unused field. Required because UniFFI can't handle variants without fields.
unused: i8,
},
/// An unrecoverable upload failure.
///
/// A possible cause might be a malformed URL.
UnrecoverableFailure,
UnrecoverableFailure {
#[doc(hidden)]
/// Unused field. Required because UniFFI can't handle variants without fields.
unused: i8,
},
/// A HTTP response code.
///
/// This can still indicate an error, depending on the status code.
HttpStatus(u32),
HttpStatus {
/// The HTTP status code
code: i32,
},
}
impl From<u32> for UploadResult {
@ -54,13 +65,13 @@ impl From<u32> for UploadResult {
status if (status & UPLOAD_RESULT_HTTP_STATUS) == UPLOAD_RESULT_HTTP_STATUS => {
// Extract the status code from the lower bits.
let http_status = status & !UPLOAD_RESULT_HTTP_STATUS;
UploadResult::HttpStatus(http_status)
UploadResult::http_status(http_status as i32)
}
UPLOAD_RESULT_RECOVERABLE => UploadResult::RecoverableFailure,
UPLOAD_RESULT_UNRECOVERABLE => UploadResult::UnrecoverableFailure,
UPLOAD_RESULT_RECOVERABLE => UploadResult::recoverable_failure(),
UPLOAD_RESULT_UNRECOVERABLE => UploadResult::unrecoverable_failure(),
// Any unknown result code is treated as unrecoverable.
_ => UploadResult::UnrecoverableFailure,
_ => UploadResult::unrecoverable_failure(),
}
}
}
@ -72,12 +83,24 @@ impl UploadResult {
/// Failures are recorded in the `ping_upload_failure` metric.
pub fn get_label(&self) -> Option<&str> {
match self {
UploadResult::HttpStatus(200..=299) => None,
UploadResult::HttpStatus(400..=499) => Some("status_code_4xx"),
UploadResult::HttpStatus(500..=599) => Some("status_code_5xx"),
UploadResult::HttpStatus(_) => Some("status_code_unknown"),
UploadResult::UnrecoverableFailure => Some("unrecoverable"),
UploadResult::RecoverableFailure => Some("recoverable"),
UploadResult::HttpStatus { code: 200..=299 } => None,
UploadResult::HttpStatus { code: 400..=499 } => Some("status_code_4xx"),
UploadResult::HttpStatus { code: 500..=599 } => Some("status_code_5xx"),
UploadResult::HttpStatus { .. } => Some("status_code_unknown"),
UploadResult::UnrecoverableFailure { .. } => Some("unrecoverable"),
UploadResult::RecoverableFailure { .. } => Some("recoverable"),
}
}
pub(crate) fn recoverable_failure() -> Self {
Self::RecoverableFailure { unused: 0 }
}
pub(crate) fn unrecoverable_failure() -> Self {
Self::UnrecoverableFailure { unused: 0 }
}
pub(crate) fn http_status(code: i32) -> Self {
Self::HttpStatus { code }
}
}