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

SolrIndexTask::setIndex()   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
c 0
b 0
f 0
nc 1
nop 1
dl 0
loc 3
ccs 2
cts 2
cp 1
crap 1
rs 10
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
    /**
75
     * Default batch length
76
     *
77
     * @var int
78
     */
79
    protected $batchLength = 1;
80
81
    /**
82
     * @var BaseIndex Current core being indexed
83
     */
84
    protected $index;
85
86
    /**
87
     * SolrIndexTask constructor. Sets up the document factory
88
     *
89
     * @throws ReflectionException
90
     */
91 15
    public function __construct()
92
    {
93 15
        parent::__construct();
94
        // Only index live items.
95
        // The old FTS module also indexed Draft items. This is unnecessary
96 15
        Versioned::set_reading_mode(Versioned::DEFAULT_MODE);
97
        // If versioned is needed, a separate Versioned Search module is required
98 15
        $this->setService(Injector::inst()->get(SolrCoreService::class));
99 15
        $this->setLogger(Injector::inst()->get(LoggerInterface::class));
100 15
        $this->setDebug(Director::isDev() || Director::is_cli());
101 15
        $currentStates = SiteState::currentStates();
102 15
        SiteState::setDefaultStates($currentStates);
103 15
    }
104
105
    /**
106
     * Set the {@link SolrCoreService}
107
     *
108
     * @param SolrCoreService $service
109
     * @return SolrIndexTask
110
     */
111 15
    public function setService(SolrCoreService $service): SolrIndexTask
112
    {
113 15
        $this->service = $service;
114
115 15
        return $this;
116
    }
117
118
    /**
119
     * Set the debug mode
120
     *
121
     * @param bool $debug
122
     * @return SolrIndexTask
123
     */
124 15
    public function setDebug(bool $debug): SolrIndexTask
125
    {
126 15
        $this->debug = $debug;
127
128 15
        return $this;
129
    }
130
131
    /**
132
     * Implement this method in the task subclass to
133
     * execute via the TaskRunner
134
     *
135
     * @param HTTPRequest $request
136
     * @return int|bool
137
     * @throws Exception
138
     * @throws GuzzleException
139
     */
140 13
    public function run($request)
141
    {
142 13
        $start = time();
143 13
        $this->getLogger()->info(date('Y-m-d H:i:s'));
144 13
        list($vars, $group, $isGroup) = $this->taskSetup($request);
145 13
        $groups = 0;
146 13
        $indexes = $this->service->getValidIndexes($request->getVar('index'));
147
148 13
        foreach ($indexes as $indexName) {
149
            /** @var BaseIndex $index */
150 13
            $index = Injector::inst()->get($indexName, false);
151 13
            $this->setIndex($index);
152
153 13
            $indexClasses = $this->index->getClasses();
154 13
            $classes = $this->getClasses($vars, $indexClasses);
155 13
            if (!count($classes)) {
156 10
                continue;
157
            }
158
159 13
            $this->clearIndex($vars);
160
161 13
            $groups = $this->indexClassForIndex($classes, $isGroup, $group);
162
        }
163
164 13
        $this->getLogger()->info(date('Y-m-d H:i:s'));
165 13
        $this->getLogger()->info(sprintf('Time taken: %s minutes', (time() - $start) / 60));
166
167 13
        return $groups;
168
    }
169
170
    /**
171
     * Set up the requirements for this task
172
     *
173
     * @param HTTPRequest $request
174
     * @return array
175
     */
176 13
    protected function taskSetup($request): array
177
    {
178 13
        $vars = $request->getVars();
179 13
        $this->debug = $this->debug || isset($vars['debug']);
180 13
        $group = $vars['group'] ?? 0;
181 13
        $start = $vars['start'] ?? 0;
182 13
        $group = ($start > $group) ? $start : $group;
183 13
        $isGroup = isset($vars['group']);
184
185 13
        return [$vars, $group, $isGroup];
186
    }
187
188
    /**
189
     * get the classes to run for this task execution
190
     *
191
     * @param $vars
192
     * @param array $classes
193
     * @return bool|array
194
     */
195 13
    protected function getClasses($vars, array $classes): array
196
    {
197 13
        if (isset($vars['class'])) {
198 1
            return array_intersect($classes, [$vars['class']]);
199
        }
200
201 12
        return $classes;
202
    }
203
204
    /**
205
     * Clear the given index if a full re-index is needed
206
     *
207
     * @param $vars
208
     * @throws Exception
209
     */
210 13
    public function clearIndex($vars)
211
    {
212 13
        if (!empty($vars['clear'])) {
213 1
            $this->getLogger()->info(sprintf('Clearing index %s', $this->index->getIndexName()));
214 1
            $this->service->doManipulate(ArrayList::create([]), SolrCoreService::DELETE_TYPE_ALL, $this->index);
215
        }
216 13
    }
217
218
    /**
219
     * Index the classes for a specific index
220
     *
221
     * @param $classes
222
     * @param $isGroup
223
     * @param $group
224
     * @return int
225
     * @throws Exception
226
     * @throws GuzzleException
227
     */
228 13
    protected function indexClassForIndex($classes, $isGroup, $group): int
229
    {
230 13
        $groups = 0;
231 13
        foreach ($classes as $class) {
232 13
            $groups = $this->indexClass($isGroup, $class, $group);
233
        }
234
235 13
        return $groups;
236
    }
237
238
    /**
239
     * Index a single class for a given index. {@link static::indexClassForIndex()}
240
     *
241
     * @param bool $isGroup
242
     * @param string $class
243
     * @param BaseIndex $index
244
     * @param int $group
245
     * @return int
246
     * @throws GuzzleException
247
     * @throws ValidationException
248
     */
249 13
    private function indexClass($isGroup, $class, int $group): int
250
    {
251 13
        $index = $this->getIndex();
252 13
        $this->getLogger()->info(sprintf('Indexing %s for %s', $class, $index->getIndexName()));
253 13
        $this->batchLength = DocumentFactory::config()->get('batchLength');
254 13
        $totalGroups = (int)ceil($class::get()->count() / $this->batchLength);
255 13
        $cores = SolrCoreService::config()->get('cpucores') ?: 1;
256 13
        $groups = $isGroup ? ($group + $cores - 1) : $totalGroups;
257 13
        $this->getLogger()->info(sprintf('Total groups %s', $totalGroups));
258
        do { // Run from oldest to newest
259
            try {
260
                // The unittest param is from phpunit.xml.dist, meant to bypass the exit(0) call
261 13
                if (function_exists('pcntl_fork') &&
262 13
                    !Controller::curr()->getRequest()->getVar('unittest')
263
                ) {
264
                    $group = $this->spawnChildren($class, $group, $cores, $groups);
265
                } else {
266 13
                    $this->doReindex($group, $class);
267
                }
268
            } catch (Exception $error) {
269
                $this->logException($index->getIndexName(), $group, $error);
270
                $group++;
271
                continue;
272
            }
273 13
            $group++;
274 13
        } while ($group <= $groups);
275
276 13
        return $totalGroups;
277
    }
278
279
    /**
280
     * @return BaseIndex
281
     */
282 14
    public function getIndex(): BaseIndex
283
    {
284 14
        return $this->index;
285
    }
286
287
    /**
288
     * @param BaseIndex $index
289
     */
290 14
    public function setIndex(BaseIndex $index): void
291
    {
292 14
        $this->index = $index;
293 14
    }
294
295
    /**
296
     * For each core, spawn a child process that will handle a separate group.
297
     * This speeds up indexing through CLI massively.
298
     *
299
     * @param string $class
300
     * @param int $group
301
     * @param int $cores
302
     * @param int $groups
303
     * @return int
304
     * @throws Exception
305
     * @throws GuzzleException
306
     */
307
    private function spawnChildren($class, int $group, int $cores, int $groups): int
308
    {
309
        $start = $group;
310
        $pids = [];
311
        // for each core, start a grouped indexing
312
        for ($i = 0; $i < $cores; $i++) {
313
            $start = $group + $i;
314
            if ($start < $groups) {
315
                $this->runForkedChild($class, $pids, $i, $start);
316
            }
317
        }
318
        // Wait for each child to finish
319
        foreach ($pids as $key => $pid) {
320
            pcntl_waitpid($pid, $status);
321
            if ($status === 0) {
322
                unset($pids[$key]);
323
            }
324
        }
325
        $commit = $this->index->getClient()->createUpdate();
326
        $commit->addCommit();
327
328
        $this->index->getClient()->update($commit);
329
330
        return $start;
331
    }
332
333
    /**
334
     * Create a fork and run the child
335
     *
336
     * @param string $class Class to index
337
     * @param BaseIndex $index Index to push the docs to
338
     * @param array $pids Array of all the child PID's
339
     * @param int $coreNumber Core number
340
     * @param int $start Start point for the objects
341
     * @return void
342
     * @throws GuzzleException
343
     * @throws ValidationException
344
     */
345
    private function runForkedChild($class, array &$pids, int $coreNumber, int $start): void
346
    {
347
        $pid = pcntl_fork();
348
        // PID needs to be pushed before anything else, for some reason
349
        $pids[$coreNumber] = $pid;
350
        $config = DB::getConfig();
351
        DB::connect($config);
352
        $this->runChild($class, $pid, $start);
353
    }
354
355
    /**
356
     * Ren a single child index operation
357
     *
358
     * @param string $class Class to index
359
     * @param int $pid PID of the child
360
     * @param int $start Position to start
361
     * @throws GuzzleException
362
     * @throws ValidationException
363
     * @throws Exception
364
     */
365
    private function runChild($class, int $pid, int $start): void
366
    {
367
        if (!$pid) {
368
            try {
369
                $this->doReindex($start, $class, true);
370
            } catch (Exception $e) {
371
                SolrLogger::logMessage('ERROR', $e, $this->index->getIndexName());
372
                $msg = sprintf(
373
                    'Something went wrong while indexing %s on %s, see the logs for details',
374
                    $start,
375
                    $this->index->getIndexName()
376
                );
377
                throw new Exception($msg);
378
            }
379
        }
380
    }
381
382
    /**
383
     * Reindex the given group, for each state
384
     *
385
     * @param int $group
386
     * @param string $class
387
     * @param bool $pcntl
388
     * @throws Exception
389
     */
390 13
    private function doReindex($group, $class, $pcntl = false)
391
    {
392 13
        foreach (SiteState::getStates() as $state) {
393 13
            if ($state !== 'default' && !empty($state)) {
394
                SiteState::withState($state);
395
            }
396 13
            $this->stateReindex($group, $class);
397
        }
398
399 13
        SiteState::withState(SiteState::DEFAULT_STATE);
400 13
        $this->getLogger()->info(sprintf('Indexed group %s', $group));
401
402 13
        if ($pcntl) {
403
            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...
404
        }
405 13
    }
406
407
    /**
408
     * Index a group of a class for a specific state and index
409
     *
410
     * @param $group
411
     * @param $class
412
     * @throws Exception
413
     */
414 13
    private function stateReindex($group, $class): void
415
    {
416
        // Generate filtered list of local records
417 13
        $baseClass = DataObject::getSchema()->baseDataClass($class);
418
        /** @var DataList|DataObject[] $items */
419 13
        $items = DataObject::get($baseClass)
420 13
            ->sort('ID ASC')
421 13
            ->limit($this->batchLength, ($group * $this->batchLength));
422 13
        if ($items->count()) {
423 1
            $this->updateIndex($items);
424
        }
425 13
    }
426
427
    /**
428
     * Execute the update on the client
429
     *
430
     * @param $items
431
     * @throws Exception
432
     */
433 1
    private function updateIndex($items): void
434
    {
435 1
        $index = $this->getIndex();
436 1
        $client = $index->getClient();
437 1
        $update = $client->createUpdate();
438 1
        $this->service->setInDebugMode($this->debug);
439 1
        $this->service->updateIndex($index, $items, $update);
440 1
        $client->update($update);
441 1
    }
442
443
    /**
444
     * Log an exception if it happens. Most are catched, these logs are for the developers
445
     * to identify problems and fix them.
446
     *
447
     * @param string $index
448
     * @param int $group
449
     * @param Exception $exception
450
     * @throws GuzzleException
451
     * @throws ValidationException
452
     */
453
    private function logException($index, int $group, Exception $exception): void
454
    {
455
        $this->getLogger()->error($exception->getMessage());
456
        $msg = sprintf(
457
            'Error indexing core %s on group %s,' . PHP_EOL .
458
            'Please log in to the CMS to find out more about Indexing errors' . PHP_EOL,
459
            $index,
460
            $group
461
        );
462
        SolrLogger::logMessage('ERROR', $msg, $index);
463
    }
464
}
465