Passed
Push — hans/code-cleanup ( 9ac287...d8c8d6 )
by Simon
05:51
created

SolrIndexTask::setCores()   A

Complexity

Conditions 1
Paths 1

Size

Total Lines 3
Code Lines 1

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 2
CRAP Score 1

Importance

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

In general, usage of exit should be done with care and only when running in a scripting context like a CLI script.

Loading history...
407
        }
408 13
    }
409
410
    /**
411
     * Index a group of a class for a specific state and index
412
     *
413
     * @param $group
414
     * @param $class
415
     * @throws Exception
416
     */
417 13
    private function stateReindex($group, $class): void
418
    {
419
        // Generate filtered list of local records
420 13
        $baseClass = DataObject::getSchema()->baseDataClass($class);
421
        /** @var DataList|DataObject[] $items */
422 13
        $items = DataObject::get($baseClass)
423 13
            ->sort('ID ASC')
424 13
            ->limit($this->batchLength, ($group * $this->batchLength));
425 13
        if ($items->count()) {
426 1
            $this->updateIndex($items);
427
        }
428 13
    }
429
430
    /**
431
     * Execute the update on the client
432
     *
433
     * @param $items
434
     * @throws Exception
435
     */
436 1
    private function updateIndex($items): void
437
    {
438 1
        $index = $this->getIndex();
439 1
        $client = $index->getClient();
440 1
        $update = $client->createUpdate();
441 1
        $this->service->setInDebugMode($this->debug);
442 1
        $this->service->updateIndex($index, $items, $update);
443 1
        $client->update($update);
444 1
    }
445
446
    /**
447
     * Log an exception if it happens. Most are catched, these logs are for the developers
448
     * to identify problems and fix them.
449
     *
450
     * @param string $index
451
     * @param int $group
452
     * @param Exception $exception
453
     * @throws GuzzleException
454
     * @throws ValidationException
455
     */
456
    private function logException($index, int $group, Exception $exception): void
457
    {
458
        $this->getLogger()->error($exception->getMessage());
459
        $msg = sprintf(
460
            'Error indexing core %s on group %s,' . PHP_EOL .
461
            'Please log in to the CMS to find out more about Indexing errors' . PHP_EOL,
462
            $index,
463
            $group
464
        );
465
        SolrLogger::logMessage('ERROR', $msg, $index);
466
    }
467
468
    /**
469
     * @return int
470
     */
471 1
    public function getCores(): int
472
    {
473 1
        return $this->cores;
474
    }
475
476
    /**
477
     * @param int $cores
478
     */
479 15
    public function setCores(int $cores): void
480
    {
481 15
        $this->cores = $cores;
482 15
    }
483
484
    /**
485
     * @return int
486
     */
487
    public function getBatchLength(): int
488
    {
489
        return $this->batchLength;
490
    }
491
492
    /**
493
     * @param int $batchLength
494
     */
495 15
    public function setBatchLength(int $batchLength): void
496
    {
497 15
        $this->batchLength = $batchLength;
498 15
    }
499
}
500