Passed
Push — hans/code-cleanup ( 74c548...547d18 )
by Simon
09:31 queued 07:32
created

SolrIndexTask::doReindex()   A

Complexity

Conditions 6
Paths 9

Size

Total Lines 16
Code Lines 10

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 8
CRAP Score 6.73

Importance

Changes 11
Bugs 2 Features 0
Metric Value
cc 6
eloc 10
nc 9
nop 3
dl 0
loc 16
ccs 8
cts 11
cp 0.7272
crap 6.73
rs 9.2222
c 11
b 2
f 0
1
<?php
2
3
4
namespace Firesphere\SolrSearch\Tasks;
5
6
use Exception;
7
use Firesphere\SolrSearch\Factories\DocumentFactory;
8
use Firesphere\SolrSearch\Helpers\SolrLogger;
9
use Firesphere\SolrSearch\Indexes\BaseIndex;
10
use Firesphere\SolrSearch\Services\SolrCoreService;
11
use Firesphere\SolrSearch\States\SiteState;
12
use Firesphere\SolrSearch\Traits\LoggerTrait;
13
use Firesphere\SolrSearch\Traits\SolrIndexTrait;
14
use GuzzleHttp\Exception\GuzzleException;
15
use Psr\Log\LoggerInterface;
16
use ReflectionException;
17
use SilverStripe\Control\Controller;
18
use SilverStripe\Control\Director;
19
use SilverStripe\Control\HTTPRequest;
20
use SilverStripe\Core\Injector\Injector;
21
use SilverStripe\Dev\BuildTask;
22
use SilverStripe\ORM\ArrayList;
23
use SilverStripe\ORM\DataList;
24
use SilverStripe\ORM\DataObject;
25
use SilverStripe\ORM\DB;
26
use SilverStripe\ORM\SS_List;
27
use SilverStripe\ORM\ValidationException;
28
use SilverStripe\Versioned\Versioned;
29
30
/**
31
 * Class SolrIndexTask
32
 *
33
 * @description Index items to Solr through a tasks
34
 * @package Firesphere\SolrSearch\Tasks
35
 */
36
class SolrIndexTask extends BuildTask
37
{
38
    use LoggerTrait;
39
    use SolrIndexTrait;
40
    /**
41
     * URLSegment of this task
42
     *
43
     * @var string
44
     */
45
    private static $segment = 'SolrIndexTask';
46
    /**
47
     * Store the current states for all instances of SiteState
48
     *
49
     * @var array
50
     */
51
    public $currentStates;
52
    /**
53
     * My name
54
     *
55
     * @var string
56
     */
57
    protected $title = 'Solr Index update';
58
    /**
59
     * What do I do?
60
     *
61
     * @var string
62
     */
63
    protected $description = 'Add or update documents to an existing Solr core.';
64
65
    /**
66
     * SolrIndexTask constructor. Sets up the document factory
67
     *
68
     * @throws ReflectionException
69
     */
70 15
    public function __construct()
71
    {
72 15
        parent::__construct();
73
        // Only index live items.
74
        // The old FTS module also indexed Draft items. This is unnecessary
75 15
        Versioned::set_reading_mode(Versioned::DEFAULT_MODE);
76
        // If versioned is needed, a separate Versioned Search module is required
77 15
        $this->setService(Injector::inst()->get(SolrCoreService::class));
78 15
        $this->setLogger(Injector::inst()->get(LoggerInterface::class));
79 15
        $this->setDebug(Director::isDev() || Director::is_cli());
80 15
        $this->setBatchLength(DocumentFactory::config()->get('batchLength'));
81 15
        $cores = SolrCoreService::config()->get('cpucores') ?: 1;
82 15
        $this->setCores($cores);
83 15
        $currentStates = SiteState::currentStates();
84 15
        SiteState::setDefaultStates($currentStates);
85 15
    }
86
87
    /**
88
     * Implement this method in the task subclass to
89
     * execute via the TaskRunner
90
     *
91
     * @param HTTPRequest $request Current request
92
     * @return int|bool
93
     * @throws Exception
94
     * @throws GuzzleException
95
     */
96 13
    public function run($request)
97
    {
98 13
        $start = time();
99 13
        $this->getLogger()->info(date('Y-m-d H:i:s'));
100 13
        list($vars, $group, $isGroup) = $this->taskSetup($request);
101 13
        $groups = 0;
102 13
        $indexes = $this->service->getValidIndexes($request->getVar('index'));
103
104 13
        foreach ($indexes as $indexName) {
105
            /** @var BaseIndex $index */
106 13
            $index = Injector::inst()->get($indexName, false);
107 13
            $this->setIndex($index);
108
109 13
            $indexClasses = $this->index->getClasses();
110 13
            $classes = $this->getClasses($vars, $indexClasses);
111 13
            if (!count($classes)) {
112 10
                continue;
113
            }
114
115 13
            $this->clearIndex($vars);
116
117 13
            $groups = $this->indexClassForIndex($classes, $isGroup, $group);
118
        }
119
120 13
        $this->getLogger()->info(date('Y-m-d H:i:s'));
121 13
        $this->getLogger()->info(sprintf('Time taken: %s minutes', (time() - $start) / 60));
122
123 13
        return $groups;
124
    }
125
126
    /**
127
     * Set up the requirements for this task
128
     *
129
     * @param HTTPRequest $request Current request
130
     * @return array
131
     */
132 13
    protected function taskSetup($request): array
133
    {
134 13
        $vars = $request->getVars();
135 13
        $this->debug = $this->debug || isset($vars['debug']);
136 13
        $group = $vars['group'] ?? 0;
137 13
        $start = $vars['start'] ?? 0;
138 13
        $group = ($start > $group) ? $start : $group;
139 13
        $isGroup = isset($vars['group']);
140
141 13
        return [$vars, $group, $isGroup];
142
    }
143
144
    /**
145
     * get the classes to run for this task execution
146
     *
147
     * @param array $vars URL GET Parameters
148
     * @param array $classes Classes to index
149
     * @return bool|array
150
     */
151 13
    protected function getClasses($vars, array $classes): array
152
    {
153 13
        if (isset($vars['class'])) {
154 1
            return array_intersect($classes, [$vars['class']]);
155
        }
156
157 12
        return $classes;
158
    }
159
160
    /**
161
     * Clear the given index if a full re-index is needed
162
     *
163
     * @param array $vars URL GET Parameters
164
     * @throws Exception
165
     */
166 13
    public function clearIndex($vars)
167
    {
168 13
        if (!empty($vars['clear'])) {
169 1
            $this->getLogger()->info(sprintf('Clearing index %s', $this->index->getIndexName()));
170 1
            $this->service->doManipulate(ArrayList::create([]), SolrCoreService::DELETE_TYPE_ALL, $this->index);
171
        }
172 13
    }
173
174
    /**
175
     * Index the classes for a specific index
176
     *
177
     * @param array $classes Classes that need indexing
178
     * @param bool $isGroup Indexing a specific group?
179
     * @param int $group Group to index
180
     * @return int
181
     * @throws Exception
182
     * @throws GuzzleException
183
     */
184 13
    protected function indexClassForIndex($classes, $isGroup, $group): int
185
    {
186 13
        $groups = 0;
187 13
        foreach ($classes as $class) {
188 13
            $groups = $this->indexClass($isGroup, $class, $group);
189
        }
190
191 13
        return $groups;
192
    }
193
194
    /**
195
     * Index a single class for a given index. {@link static::indexClassForIndex()}
196
     *
197
     * @param bool $isGroup Is a specific group indexed
198
     * @param string $class Class to index
199
     * @param int $group Group to index
200
     * @return int
201
     * @throws GuzzleException
202
     * @throws ValidationException
203
     */
204 13
    private function indexClass($isGroup, $class, int $group): int
205
    {
206 13
        $this->getLogger()->info(sprintf('Indexing %s for %s', $class, $this->getIndex()->getIndexName()));
207 13
        list($totalGroups, $groups) = $this->getGroupSettings($isGroup, $class, $group);
208 13
        $this->getLogger()->info(sprintf('Total groups %s', $totalGroups));
209
        do { // Run from oldest to newest
210
            try {
211
                // The unittest param is from phpunit.xml.dist, meant to bypass the exit(0) call
212
                // The pcntl check is for unit tests, but CircleCI does not support PCNTL (yet)
213
                // @todo fix CircleCI to support PCNTL
214 13
                if ($this->hasPCNTL()) {
215
                    $group = $this->spawnChildren($class, $group, $groups);
216
                } else {
217 13
                    $this->doReindex($group, $class);
218
                }
219 13
                $group++;
220
            } catch (Exception $error) {
221
                $this->logException($this->index->getIndexName(), $group, $error);
222
                $group++;
223
                continue;
224
            }
225 13
        } while ($group <= $groups);
226
227 13
        return $totalGroups;
228
    }
229
230
    /**
231
     * Check the amount of groups and the total against the isGroup check.
232
     *
233
     * @param bool $isGroup Is it a specific group
234
     * @param string $class Class to check
235
     * @param int $group Current group to index
236
     * @return array
237
     */
238 13
    private function getGroupSettings($isGroup, $class, int $group): array
239
    {
240 13
        $totalGroups = (int)ceil($class::get()->count() / $this->getBatchLength());
241 13
        $groups = $isGroup ? ($group + $this->getCores() - 1) : $totalGroups;
242
243 13
        return [$totalGroups, $groups];
244
    }
245
246 13
    private function hasPCNTL()
247
    {
248 13
        return function_exists('pcntl_fork') &&
249
            (Controller::curr()->getRequest()->getVar('unittest') === 'pcntl') ||
250 13
            !Controller::curr()->getRequest()->getVar('unittest');
251
    }
252
253
    /**
254
     * For each core, spawn a child process that will handle a separate group.
255
     * This speeds up indexing through CLI massively.
256
     *
257
     * @param string $class Class to index
258
     * @param int $group Group to index
259
     * @param int $groups Total amount of groups
260
     * @return int Last group indexed
261
     * @throws Exception
262
     * @throws GuzzleException
263
     */
264
    private function spawnChildren($class, int $group, int $groups): int
265
    {
266
        $start = $group;
267
        $pids = [];
268
        $cores = $this->getCores();
269
        // for each core, start a grouped indexing
270
        for ($i = 0; $i < $cores; $i++) {
271
            $start = $group + $i;
272
            if ($start < $groups) {
273
                $this->runForkedChild($class, $pids, $i, $start);
274
            }
275
        }
276
        // Wait for each child to finish
277
        foreach ($pids as $key => $pid) {
278
            pcntl_waitpid($pid, $status);
279
            if ($status === 0) {
280
                unset($pids[$key]);
281
            }
282
        }
283
        $commit = $this->index->getClient()->createUpdate();
284
        $commit->addCommit();
285
286
        $this->index->getClient()->update($commit);
287
288
        return $start;
289
    }
290
291
    /**
292
     * Create a fork and run the child
293
     *
294
     * @param string $class Class to index
295
     * @param array $pids Array of all the child Process IDs
296
     * @param int $coreNumber Core number
297
     * @param int $start Start point for the objects
298
     * @return void
299
     * @throws GuzzleException
300
     * @throws ValidationException
301
     */
302
    private function runForkedChild($class, array &$pids, int $coreNumber, int $start): void
303
    {
304
        $pid = pcntl_fork();
305
        // PID needs to be pushed before anything else, for some reason
306
        $pids[$coreNumber] = $pid;
307
        $config = DB::getConfig();
308
        DB::connect($config);
309
        $this->runChild($class, $pid, $start);
310
    }
311
312
    /**
313
     * Ren a single child index operation
314
     *
315
     * @param string $class Class to index
316
     * @param int $pid PID of the child
317
     * @param int $start Position to start
318
     * @throws GuzzleException
319
     * @throws ValidationException
320
     * @throws Exception
321
     */
322
    private function runChild($class, int $pid, int $start): void
323
    {
324
        if (!$pid) {
325
            try {
326
                $this->doReindex($start, $class, $pid);
327
            } catch (Exception $e) {
328
                SolrLogger::logMessage('ERROR', $e, $this->index->getIndexName());
329
                $msg = sprintf(
330
                    'Something went wrong while indexing %s on %s, see the logs for details',
331
                    $start,
332
                    $this->index->getIndexName()
333
                );
334
                throw new Exception($msg);
335
            }
336
        }
337
    }
338
339
    /**
340
     * Reindex the given group, for each state
341
     *
342
     * @param int $group Group to index
343
     * @param string $class Class to index
344
     * @param bool|int $pcntl Are we a child process or not
345
     * @throws Exception
346
     */
347 13
    private function doReindex($group, $class, $pcntl = false)
348
    {
349 13
        foreach (SiteState::getStates() as $state) {
350 13
            if ($state !== 'default' && !empty($state)) {
351
                SiteState::withState($state);
352
            }
353 13
            $this->stateReindex($group, $class);
354
        }
355
356 13
        SiteState::withState(SiteState::DEFAULT_STATE);
357 13
        $this->getLogger()->info(sprintf('Indexed group %s', $group));
358
359 13
        if (Controller::curr()->getRequest()->getVar('unittest') === 'pcntl') {
360
            throw new Exception('Catch exception for unittests');
361 13
        } elseif ($pcntl !== false) {
362
            exit(0);
363
        }
364 13
    }
365
366
    /**
367
     * Index a group of a class for a specific state and index
368
     *
369
     * @param string $group Group to index
370
     * @param string $class Class to index
371
     * @throws Exception
372
     */
373 13
    private function stateReindex($group, $class): void
374
    {
375
        // Generate filtered list of local records
376 13
        $baseClass = DataObject::getSchema()->baseDataClass($class);
377
        /** @var DataList|DataObject[] $items */
378 13
        $items = DataObject::get($baseClass)
379 13
            ->sort('ID ASC')
380 13
            ->limit($this->getBatchLength(), ($group * $this->getBatchLength()));
381 13
        if ($items->count()) {
382 1
            $this->updateIndex($items);
383
        }
384 13
    }
385
386
    /**
387
     * Execute the update on the client
388
     *
389
     * @param SS_List $items Items to index
390
     * @throws Exception
391
     */
392 1
    private function updateIndex($items): void
393
    {
394 1
        $index = $this->getIndex();
395 1
        $client = $index->getClient();
396 1
        $update = $client->createUpdate();
397 1
        $this->service->setInDebugMode($this->debug);
398 1
        $this->service->updateIndex($index, $items, $update);
399 1
        $client->update($update);
400 1
    }
401
402
    /**
403
     * Log an exception if it happens. Most are catched, these logs are for the developers
404
     * to identify problems and fix them.
405
     *
406
     * @param string $index Index that is currently running
407
     * @param int $group Group currently attempted to index
408
     * @param Exception $exception Exception that's been thrown
409
     * @throws GuzzleException
410
     * @throws ValidationException
411
     */
412
    private function logException($index, int $group, Exception $exception): void
413
    {
414
        $this->getLogger()->error($exception->getMessage());
415
        $msg = sprintf(
416
            'Error indexing core %s on group %s,' . PHP_EOL .
417
            'Please log in to the CMS to find out more about Indexing errors' . PHP_EOL,
418
            $index,
419
            $group
420
        );
421
        SolrLogger::logMessage('ERROR', $msg, $index);
422
    }
423
}
424