Passed
Push — hans/code-cleanup ( de6908...bf2e41 )
by Simon
07:25 queued 05:27
created

SolrIndexTask::indexClass()   B

Complexity

Conditions 7
Paths 10

Size

Total Lines 27
Code Lines 20

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 13
CRAP Score 8.0504

Importance

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