1 /*
2  * Copyright (C) 2022 The Android Open Source Project
3  *
4  * Licensed under the Apache License, Version 2.0 (the "License");
5  * you may not use this file except in compliance with the License.
6  * You may obtain a copy of the License at
7  *
8  *      http://www.apache.org/licenses/LICENSE-2.0
9  *
10  * Unless required by applicable law or agreed to in writing, software
11  * distributed under the License is distributed on an "AS IS" BASIS,
12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13  * See the License for the specific language governing permissions and
14  * limitations under the License.
15  */
16 
17 package com.android.adservices.service.measurement.reporting;
18 
19 import static com.android.adservices.service.measurement.util.JobLockHolder.Type.AGGREGATE_REPORTING;
20 import static com.android.adservices.service.stats.AdServicesStatsLog.AD_SERVICES_BACKGROUND_JOBS_EXECUTION_REPORTED__EXECUTION_RESULT_CODE__SKIP_FOR_KILL_SWITCH_ON;
21 import static com.android.adservices.spe.AdServicesJobInfo.MEASUREMENT_AGGREGATE_MAIN_REPORTING_JOB;
22 
23 import android.app.job.JobInfo;
24 import android.app.job.JobParameters;
25 import android.app.job.JobScheduler;
26 import android.app.job.JobService;
27 import android.content.ComponentName;
28 import android.content.Context;
29 
30 import com.android.adservices.LogUtil;
31 import com.android.adservices.LoggerFactory;
32 import com.android.adservices.concurrency.AdServicesExecutors;
33 import com.android.adservices.data.measurement.DatastoreManager;
34 import com.android.adservices.data.measurement.DatastoreManagerFactory;
35 import com.android.adservices.service.Flags;
36 import com.android.adservices.service.FlagsFactory;
37 import com.android.adservices.service.common.compat.ServiceCompatUtils;
38 import com.android.adservices.service.measurement.aggregation.AggregateEncryptionKeyManager;
39 import com.android.adservices.service.measurement.util.JobLockHolder;
40 import com.android.adservices.service.stats.AdServicesLoggerImpl;
41 import com.android.adservices.spe.AdServicesJobServiceLogger;
42 import com.android.internal.annotations.VisibleForTesting;
43 
44 import com.google.common.util.concurrent.ListeningExecutorService;
45 
46 import java.util.concurrent.Future;
47 
48 /**
49  * Main service for scheduling aggregate reporting jobs. The actual job execution logic is part of
50  * {@link AggregateReportingJobHandler}
51  */
52 public final class AggregateReportingJobService extends JobService {
53     private static final int MEASUREMENT_AGGREGATE_MAIN_REPORTING_JOB_ID =
54             MEASUREMENT_AGGREGATE_MAIN_REPORTING_JOB.getJobId();
55     private static final ListeningExecutorService sBlockingExecutor =
56             AdServicesExecutors.getBlockingExecutor();
57 
58     private Future mExecutorFuture;
59 
60     @Override
onStartJob(JobParameters params)61     public boolean onStartJob(JobParameters params) {
62         // Always ensure that the first thing this job does is check if it should be running, and
63         // cancel itself if it's not supposed to be.
64         if (ServiceCompatUtils.shouldDisableExtServicesJobOnTPlus(this)) {
65             LogUtil.d(
66                     "Disabling AggregateReportingJobService job because it's running in"
67                             + " ExtServices on T+");
68             return skipAndCancelBackgroundJob(params, /* skipReason=*/ 0, /* doRecord=*/ false);
69         }
70 
71         AdServicesJobServiceLogger.getInstance()
72                 .recordOnStartJob(MEASUREMENT_AGGREGATE_MAIN_REPORTING_JOB_ID);
73 
74         if (FlagsFactory.getFlags().getMeasurementJobAggregateReportingKillSwitch()) {
75             LoggerFactory.getMeasurementLogger().e("AggregateReportingJobService is disabled");
76             return skipAndCancelBackgroundJob(
77                     params,
78                     AD_SERVICES_BACKGROUND_JOBS_EXECUTION_REPORTED__EXECUTION_RESULT_CODE__SKIP_FOR_KILL_SWITCH_ON,
79                     /* doRecord=*/ true);
80         }
81 
82         LoggerFactory.getMeasurementLogger().d("AggregateReportingJobService.onStartJob");
83         mExecutorFuture =
84                 sBlockingExecutor.submit(
85                         () -> {
86                             processPendingReports();
87 
88                             AdServicesJobServiceLogger.getInstance()
89                                     .recordJobFinished(
90                                             MEASUREMENT_AGGREGATE_MAIN_REPORTING_JOB_ID,
91                                             /* isSuccessful= */ true,
92                                             /* shouldRetry= */ false);
93 
94                             jobFinished(params, /* wantsReschedule= */ false);
95                         });
96         return true;
97     }
98 
99     @VisibleForTesting
processPendingReports()100     void processPendingReports() {
101         final JobLockHolder lock = JobLockHolder.getInstance(AGGREGATE_REPORTING);
102         if (lock.tryLock()) {
103             try {
104                 long maxAggregateReportUploadRetryWindowMs =
105                         FlagsFactory.getFlags()
106                                 .getMeasurementMaxAggregateReportUploadRetryWindowMs();
107                 DatastoreManager datastoreManager =
108                         DatastoreManagerFactory.getDatastoreManager(getApplicationContext());
109                 new AggregateReportingJobHandler(
110                                 datastoreManager,
111                                 new AggregateEncryptionKeyManager(
112                                         datastoreManager, getApplicationContext()),
113                                 FlagsFactory.getFlags(),
114                                 AdServicesLoggerImpl.getInstance(),
115                                 ReportingStatus.ReportType.AGGREGATE,
116                                 ReportingStatus.UploadMethod.REGULAR,
117                                 getApplicationContext())
118                         .performScheduledPendingReportsInWindow(
119                                 System.currentTimeMillis() - maxAggregateReportUploadRetryWindowMs,
120                                 System.currentTimeMillis());
121                 return;
122             } finally {
123                 lock.unlock();
124             }
125         }
126         LoggerFactory.getMeasurementLogger()
127                 .d("AggregateReportingJobService did not acquire the lock");
128     }
129 
130     @Override
onStopJob(JobParameters params)131     public boolean onStopJob(JobParameters params) {
132         LoggerFactory.getMeasurementLogger().d("AggregateReportingJobService.onStopJob");
133         boolean shouldRetry = true;
134         if (mExecutorFuture != null) {
135             shouldRetry = mExecutorFuture.cancel(/* mayInterruptIfRunning */ true);
136         }
137         AdServicesJobServiceLogger.getInstance()
138                 .recordOnStopJob(params, MEASUREMENT_AGGREGATE_MAIN_REPORTING_JOB_ID, shouldRetry);
139         return shouldRetry;
140     }
141 
142     /** Schedules {@link AggregateReportingJobService} */
143     @VisibleForTesting
schedule(JobScheduler jobScheduler, JobInfo job)144     static void schedule(JobScheduler jobScheduler, JobInfo job) {
145         jobScheduler.schedule(job);
146     }
147 
148     /**
149      * Schedule Aggregate Reporting Job if it is not already scheduled
150      *
151      * @param context the context
152      * @param forceSchedule flag to indicate whether to force rescheduling the job.
153      */
scheduleIfNeeded(Context context, boolean forceSchedule)154     public static void scheduleIfNeeded(Context context, boolean forceSchedule) {
155         Flags flags = FlagsFactory.getFlags();
156         if (flags.getMeasurementJobAggregateReportingKillSwitch()) {
157             LoggerFactory.getMeasurementLogger()
158                     .d("AggregateReportingJobService is disabled, skip scheduling");
159             return;
160         }
161 
162         final JobScheduler jobScheduler = context.getSystemService(JobScheduler.class);
163         if (jobScheduler == null) {
164             LoggerFactory.getMeasurementLogger().e("JobScheduler not found");
165             return;
166         }
167 
168         final JobInfo scheduledJobInfo =
169                 jobScheduler.getPendingJob(MEASUREMENT_AGGREGATE_MAIN_REPORTING_JOB_ID);
170         JobInfo jobInfo = buildJobInfo(context, flags);
171         // Schedule if it hasn't been scheduled already or force rescheduling
172         if (forceSchedule || !jobInfo.equals(scheduledJobInfo)) {
173             schedule(jobScheduler, jobInfo);
174             LoggerFactory.getMeasurementLogger().d("Scheduled AggregateReportingJobService");
175         } else {
176             LoggerFactory.getMeasurementLogger()
177                     .d("AggregateReportingJobService already scheduled, skipping reschedule");
178         }
179     }
180 
buildJobInfo(Context context, Flags flags)181     private static JobInfo buildJobInfo(Context context, Flags flags) {
182         final JobInfo job =
183                 new JobInfo.Builder(
184                                 MEASUREMENT_AGGREGATE_MAIN_REPORTING_JOB_ID,
185                                 new ComponentName(context, AggregateReportingJobService.class))
186                         .setRequiresBatteryNotLow(
187                                 flags.getMeasurementAggregateReportingJobRequiredBatteryNotLow())
188                         .setRequiredNetworkType(
189                                 flags.getMeasurementAggregateReportingJobRequiredNetworkType())
190                         .setPeriodic(flags.getMeasurementAggregateMainReportingJobPeriodMs())
191                         .setPersisted(flags.getMeasurementAggregateReportingJobPersisted())
192                         .build();
193         return job;
194     }
195 
skipAndCancelBackgroundJob( final JobParameters params, int skipReason, boolean doRecord)196     private boolean skipAndCancelBackgroundJob(
197             final JobParameters params, int skipReason, boolean doRecord) {
198         final JobScheduler jobScheduler = this.getSystemService(JobScheduler.class);
199         if (jobScheduler != null) {
200             jobScheduler.cancel(MEASUREMENT_AGGREGATE_MAIN_REPORTING_JOB_ID);
201         }
202 
203         if (doRecord) {
204             AdServicesJobServiceLogger.getInstance()
205                     .recordJobSkipped(MEASUREMENT_AGGREGATE_MAIN_REPORTING_JOB_ID, skipReason);
206         }
207 
208         // Tell the JobScheduler that the job has completed and does not need to be rescheduled.
209         jobFinished(params, false);
210 
211         // Returning false means that this job has completed its work.
212         return false;
213     }
214 
215     @VisibleForTesting
getFutureForTesting()216     Future getFutureForTesting() {
217         return mExecutorFuture;
218     }
219 }
220