Skip to content

Issue #000 fix: Remove data correction for dialCodes key #1028

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Sep 14, 2020
Merged

Issue #000 fix: Remove data correction for dialCodes key #1028

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
99b0d5f
Issue #000 fix: Remove data correction for dialCodes key
Sep 14, 2020
b5ee8b6
Issue #000 fix: Remove validation for dialCodes key
Sep 14, 2020
1e625b9
Issue #000 fix: Remove validation for dialCodes key
Sep 14, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 0 additions & 9 deletions data-pipeline-flink/druid-events-validator/src/main/resources/schemas/telemetry/filter.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,14 +2,5 @@
"type": "object",
"properties": {
},
"anyOf": [
{
"not": {
"required": [
"dialCodes"
]
}
}
],
"additionalProperties": true
}
3 changes: 2 additions & 1 deletion data-pipeline-flink/druid-events-validator/src/main/resources/schemas/telemetry/search.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,8 @@
"type": "number"
},
"filters": {
"$ref": "resource:/schemas/telemetry/filter.json"
"id": "http://api.sunbird.org/telemetry/edata/filters",
"type": "object"
},
"topn": {
"id": "http://api.sunbird.org/telemetry/edata/topn",
Expand Down
19 changes: 18 additions & 1 deletion ...ine-flink/druid-events-validator/src/test/scala/org/sunbird/dp/fixture/EventFixture.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,7 +81,7 @@ object EventFixture {
|"objecttype":"Content","mediatype":"content","status":"Live"}}
""".stripMargin

val VALID_SERACH_EVENT: String =
val VALID_SEARCH_EVENT: String =
"""
|{"eid":"SEARCH","ver":"3.0","syncts":1.59518415538E12,"ets":1.59518415538E12,"flags":
|{"pp_validation_processed":true,"pp_duplicate":false,"device_denorm":false,"dialcode_denorm":true,
Expand All @@ -97,4 +97,21 @@ object EventFixture {
|"pid":"search-service"},"did":"79838ccb0ff2c7d0a9dd05f5b337fbca","env":"search","channel":"ROOT_ORG"},
|"@version":"1","object":{"id":"KLQ2G7","type":"DialCode"}}
""".stripMargin

val SEARCH_EVENT_WITH_INCORRECT_DIALCODES_KEY: String =
"""
|{"eid":"SEARCH","ver":"3.0","syncts":1.59518415538E12,"ets":1.59518415538E12,"flags":
|{"pp_validation_processed":true,"pp_duplicate":false,"device_denorm":false,"dialcode_denorm":true,
|"content_denorm":false},"dialcodedata":{"identifier":"KLQ2G7","channel":"0123221617357783046602",
|"publisher":"MHPUBLISHER","status":2.0},"mid":"invalid_dialcode_key",
|"type":"events","tags":["kp-events"],"actor":{"id":"org.sunbird.learning.platform","type":"System"},
|"edata":{"topn":[{"identifier":"do_312528046917705728246886"}],"query":"","size":7.0,"type":"content",
|"filters":{"contentType":["TextBookUnit","Resource","TextBook","Collection","Course"],"mimeType":{},
|"resourceType":{},"status":["Live"],"objectType":["Content"],"dialCodes":"KLQ2G7","framework":{},
|"compatibilityLevel":{"max":4.0,"min":1.0},"channel":{"ne":["0124433024890224640","0124446042259128320",
|"0124487522476933120","0125840271570288640","0124453662635048969"]}},"sort":{}},
|"@timestamp":"2020-07-19T18:42:41.524Z","context":{"pdata":{"ver":"1.0","id":"prod.diksha.portal",
|"pid":"search-service"},"did":"79838ccb0ff2c7d0a9dd05f5b337fbca","env":"search","channel":"ROOT_ORG"},
|"@version":"1","object":{"id":"KLQ2G7","type":"DialCode"}}
""".stripMargin
}
37 changes: 24 additions & 13 deletions ...vents-validator/src/test/scala/org/sunbird/dp/spec/DruidValidatorStreamTaskTestSpec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,13 +13,16 @@ import org.apache.flink.streaming.api.functions.source.SourceFunction.SourceCont
import org.apache.flink.test.util.MiniClusterWithClientResource
import org.mockito.Mockito
import org.mockito.Mockito.when
import org.sunbird.dp.core.domain.EventsPath
import org.sunbird.dp.core.job.FlinkKafkaConnector
import org.sunbird.dp.fixture.EventFixture
import org.sunbird.dp.validator.domain.Event
import org.sunbird.dp.validator.task.{DruidValidatorConfig, DruidValidatorStreamTask}
import org.sunbird.dp.{BaseMetricsReporter, BaseTestSpec}
import redis.embedded.RedisServer

import scala.collection.JavaConverters._


class DruidValidatorStreamTaskTestSpec extends BaseTestSpec {

Expand Down Expand Up @@ -69,24 +72,27 @@ class DruidValidatorStreamTaskTestSpec extends BaseTestSpec {

val task = new DruidValidatorStreamTask(druidValidatorConfig, mockKafkaUtil)
task.process()
TelemetryEventsSink.values.size() should be (2)
TelemetryEventsSink.values.size() should be (3)
SummaryEventsSink.values.size() should be (1)
FailedEventsSink.values.size() should be (1)
DupEventsSink.values.size() should be (1)

val invalidDialCodeKeyEvent = TelemetryEventsSink.values.asScala.filter(event => event.mid() === "invalid_dialcode_key").head
invalidDialCodeKeyEvent.getTelemetry.read(s"${EventsPath.EDTA_FILTERS}.dialCodes") should not be None

DupEventsSink.values.get(0).getFlags.get("dv_duplicate").booleanValue() should be(true)

FailedEventsSink.values.get(0).getFlags.get("dv_processed").booleanValue() should be(false)
FailedEventsSink.values.get(0).getFlags.get("dv_validation_failed").booleanValue() should be(true)

BaseMetricsReporter.gaugeMetrics(s"${druidValidatorConfig.jobName}.${druidValidatorConfig.validationSuccessMetricsCount}").getValue() should be (3)
BaseMetricsReporter.gaugeMetrics(s"${druidValidatorConfig.jobName}.${druidValidatorConfig.validationSuccessMetricsCount}").getValue() should be (4)
BaseMetricsReporter.gaugeMetrics(s"${druidValidatorConfig.jobName}.${druidValidatorConfig.validationFailureMetricsCount}").getValue() should be (1)

BaseMetricsReporter.gaugeMetrics(s"${druidValidatorConfig.jobName}.duplicate-event-count").getValue() should be (1)
BaseMetricsReporter.gaugeMetrics(s"${druidValidatorConfig.jobName}.unique-event-count").getValue() should be (4)
BaseMetricsReporter.gaugeMetrics(s"${druidValidatorConfig.jobName}.unique-event-count").getValue() should be (5)

BaseMetricsReporter.gaugeMetrics(s"${druidValidatorConfig.jobName}.${druidValidatorConfig.summaryRouterMetricCount}").getValue() should be (1)
BaseMetricsReporter.gaugeMetrics(s"${druidValidatorConfig.jobName}.${druidValidatorConfig.telemetryRouterMetricCount}").getValue() should be (2)
BaseMetricsReporter.gaugeMetrics(s"${druidValidatorConfig.jobName}.${druidValidatorConfig.telemetryRouterMetricCount}").getValue() should be (3)

}

Expand All @@ -110,7 +116,7 @@ class DruidValidatorStreamTaskTestSpec extends BaseTestSpec {

val task = new DruidValidatorStreamTask(druidValidatorConfig, mockKafkaUtil)
task.process()
TelemetryEventsSink.values.size() should be (4)
TelemetryEventsSink.values.size() should be (5)
SummaryEventsSink.values.size() should be (1)
FailedEventsSink.values.size() should be (0)
DupEventsSink.values.size() should be (0)
Expand All @@ -122,7 +128,7 @@ class DruidValidatorStreamTaskTestSpec extends BaseTestSpec {
BaseMetricsReporter.gaugeMetrics(s"${druidValidatorConfig.jobName}.unique-event-count").getValue() should be (0)

BaseMetricsReporter.gaugeMetrics(s"${druidValidatorConfig.jobName}.${druidValidatorConfig.summaryRouterMetricCount}").getValue() should be (1)
BaseMetricsReporter.gaugeMetrics(s"${druidValidatorConfig.jobName}.${druidValidatorConfig.telemetryRouterMetricCount}").getValue() should be (4)
BaseMetricsReporter.gaugeMetrics(s"${druidValidatorConfig.jobName}.${druidValidatorConfig.telemetryRouterMetricCount}").getValue() should be (5)

}

Expand All @@ -146,19 +152,22 @@ class DruidValidatorStreamTaskTestSpec extends BaseTestSpec {

val task = new DruidValidatorStreamTask(druidValidatorConfig, mockKafkaUtil)
task.process()
TelemetryEventsSink.values.size() should be (3)
TelemetryEventsSink.values.size() should be (4)
SummaryEventsSink.values.size() should be (1)
FailedEventsSink.values.size() should be (1)
DupEventsSink.values.size() should be (0)

BaseMetricsReporter.gaugeMetrics(s"${druidValidatorConfig.jobName}.${druidValidatorConfig.validationSuccessMetricsCount}").getValue() should be (4)
val invalidDialCodeKeyEvent = TelemetryEventsSink.values.asScala.filter(event => event.mid() === "invalid_dialcode_key").head
invalidDialCodeKeyEvent.getTelemetry.read(s"${EventsPath.EDTA_FILTERS}.dialCodes") should not be None

BaseMetricsReporter.gaugeMetrics(s"${druidValidatorConfig.jobName}.${druidValidatorConfig.validationSuccessMetricsCount}").getValue() should be (5)
BaseMetricsReporter.gaugeMetrics(s"${druidValidatorConfig.jobName}.${druidValidatorConfig.validationFailureMetricsCount}").getValue() should be (1)

BaseMetricsReporter.gaugeMetrics(s"${druidValidatorConfig.jobName}.duplicate-event-count").getValue() should be (0)
BaseMetricsReporter.gaugeMetrics(s"${druidValidatorConfig.jobName}.unique-event-count").getValue() should be (0)

BaseMetricsReporter.gaugeMetrics(s"${druidValidatorConfig.jobName}.${druidValidatorConfig.summaryRouterMetricCount}").getValue() should be (1)
BaseMetricsReporter.gaugeMetrics(s"${druidValidatorConfig.jobName}.${druidValidatorConfig.telemetryRouterMetricCount}").getValue() should be (3)
BaseMetricsReporter.gaugeMetrics(s"${druidValidatorConfig.jobName}.${druidValidatorConfig.telemetryRouterMetricCount}").getValue() should be (4)

}

Expand All @@ -182,7 +191,7 @@ class DruidValidatorStreamTaskTestSpec extends BaseTestSpec {

val task = new DruidValidatorStreamTask(druidValidatorConfig, mockKafkaUtil)
task.process()
TelemetryEventsSink.values.size() should be (3)
TelemetryEventsSink.values.size() should be (4)
SummaryEventsSink.values.size() should be (1)
FailedEventsSink.values.size() should be (0)
DupEventsSink.values.size() should be (1)
Expand All @@ -191,10 +200,10 @@ class DruidValidatorStreamTaskTestSpec extends BaseTestSpec {
BaseMetricsReporter.gaugeMetrics(s"${druidValidatorConfig.jobName}.${druidValidatorConfig.validationFailureMetricsCount}").getValue() should be (0)

BaseMetricsReporter.gaugeMetrics(s"${druidValidatorConfig.jobName}.duplicate-event-count").getValue() should be (1)
BaseMetricsReporter.gaugeMetrics(s"${druidValidatorConfig.jobName}.unique-event-count").getValue() should be (4)
BaseMetricsReporter.gaugeMetrics(s"${druidValidatorConfig.jobName}.unique-event-count").getValue() should be (5)

BaseMetricsReporter.gaugeMetrics(s"${druidValidatorConfig.jobName}.${druidValidatorConfig.summaryRouterMetricCount}").getValue() should be (1)
BaseMetricsReporter.gaugeMetrics(s"${druidValidatorConfig.jobName}.${druidValidatorConfig.telemetryRouterMetricCount}").getValue() should be (3)
BaseMetricsReporter.gaugeMetrics(s"${druidValidatorConfig.jobName}.${druidValidatorConfig.telemetryRouterMetricCount}").getValue() should be (4)

}

Expand All @@ -207,11 +216,13 @@ class DruidValidatorEventSource extends SourceFunction[Event] {
val event1 = gson.fromJson(EventFixture.VALID_DENORM_TELEMETRY_EVENT, new util.LinkedHashMap[String, Any]().getClass)
val event2 = gson.fromJson(EventFixture.INVALID_DENORM_TELEMETRY_EVENT, new util.LinkedHashMap[String, Any]().getClass)
val event3 = gson.fromJson(EventFixture.VALID_DENORM_SUMMARY_EVENT, new util.LinkedHashMap[String, Any]().getClass)
val event4 = gson.fromJson(EventFixture.VALID_SERACH_EVENT, new util.LinkedHashMap[String, Any]().getClass)
val event4 = gson.fromJson(EventFixture.VALID_SEARCH_EVENT, new util.LinkedHashMap[String, Any]().getClass)
val event5 = gson.fromJson(EventFixture.SEARCH_EVENT_WITH_INCORRECT_DIALCODES_KEY, new util.LinkedHashMap[String, Any]().getClass)
ctx.collect(new Event(event1))
ctx.collect(new Event(event2))
ctx.collect(new Event(event3))
ctx.collect(new Event(event4))
ctx.collect(new Event(event5))
ctx.collect(new Event(event1))
}

Expand Down
5 changes: 1 addition & 4 deletions ...ine-preprocessor/src/main/scala/org/sunbird/dp/preprocessor/util/TelemetryValidator.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,15 +30,12 @@ class TelemetryValidator(config: PipelinePreprocessorConfig) extends java.io.Ser
}
}

private def dataCorrection(event: Event): Event = {
private def dataCorrection(event: Event): Unit = {
// Remove prefix from federated userIds
val eventActorId = event.actorId()
if (eventActorId != null && !eventActorId.isEmpty && eventActorId.startsWith("f:"))
event.updateActorId(eventActorId.substring(eventActorId.lastIndexOf(":") + 1))
if (event.eid != null && event.eid().equalsIgnoreCase("SEARCH"))
event.correctDialCodeKey()
if (event.objectFieldsPresent && (event.objectType().equalsIgnoreCase("DialCode") || event.objectType().equalsIgnoreCase("qr"))) event.correctDialCodeValue()
event
}

def onValidationSuccess(event: Event, metrics: Metrics, context: ProcessFunction[Event, Event]#Context): Unit = {
Expand Down