MAGETWO-95275: Varnish "Connection reset by peer" error when large catalog is reindexed on schedule

Author: slopukhov

app/code/Magento/CacheInvalidate/Model/PurgeCache.php

@@ -26,6 +26,18 @@ class PurgeCache
      */
     private $logger;
 
+    /**
+     * Batch size of the purge request.
+     *
+     * Based on default Varnish 4 http_req_hdr_len size minus a 512 bytes margin for method,
+     * header name, line feeds etc.
+     *
+     * @see https://varnish-cache.org/docs/4.1/reference/varnishd.html
+     *
+     * @var int
+     */
+    private $requestSize = 7680;
+
     /**
      * Constructor
      *
@@ -52,10 +64,59 @@ public function __construct(
      */
     public function sendPurgeRequest($tagsPattern)
     {
+        $successful = true;
         $socketAdapter = $this->socketAdapterFactory->create();
         $servers = $this->cacheServer->getUris();
-        $headers = [self::HEADER_X_MAGENTO_TAGS_PATTERN => $tagsPattern];
         $socketAdapter->setOptions(['timeout' => 10]);
+
+        $formattedTagsChunks = $this->splitTags($tagsPattern);
+        foreach ($formattedTagsChunks as $formattedTagsChunk) {
+            if (!$this->sendPurgeRequestToServers($socketAdapter, $servers, $formattedTagsChunk)) {
+                $successful = false;
+            }
+        }
+
+        return $successful;
+    }
+
+    /**
+     * Split tags by batches
+     *
+     * @param string $tagsPattern
+     * @return \Generator
+     */
+    private function splitTags($tagsPattern)
+    {
+        $tagsBatchSize = 0;
+        $formattedTagsChunk = [];
+        $formattedTags = explode('|', $tagsPattern);
+        foreach ($formattedTags as $formattedTag) {
+            if ($tagsBatchSize + strlen($formattedTag) > $this->requestSize - count($formattedTagsChunk) - 1) {
+                yield implode('|', array_unique($formattedTagsChunk));
+                $formattedTagsChunk = [];
+                $tagsBatchSize = 0;
+            }
+
+            $tagsBatchSize += strlen($formattedTag);
+            $formattedTagsChunk[] = $formattedTag;
+        }
+        if (!empty($formattedTagsChunk)) {
+            yield implode('|', array_unique($formattedTagsChunk));
+        }
+    }
+
+    /**
+     * Send curl purge request to servers
+     * to invalidate cache by tags pattern
+     *
+     * @param \Zend\Http\Client\Adapter\Socket $socketAdapter
+     * @param \Zend\Uri\Uri[] $servers
+     * @param string $formattedTagsChunk
+     * @return bool Return true if successful; otherwise return false
+     */
+    private function sendPurgeRequestToServers($socketAdapter, $servers, $formattedTagsChunk)
+    {
+        $headers = [self::HEADER_X_MAGENTO_TAGS_PATTERN => $formattedTagsChunk];
         foreach ($servers as $server) {
             $headers['Host'] = $server->getHost();
             try {
@@ -69,12 +130,11 @@ public function sendPurgeRequest($tagsPattern)
                 $socketAdapter->read();
                 $socketAdapter->close();
             } catch (\Exception $e) {
-                $this->logger->critical($e->getMessage(), compact('server', 'tagsPattern'));
+                $this->logger->critical($e->getMessage(), compact('server', 'formattedTagsChunk'));
                 return false;
             }
         }
-
-        $this->logger->execute(compact('servers', 'tagsPattern'));
+        $this->logger->execute(compact('servers', 'formattedTagsChunk'));
         return true;
     }
 }

app/code/Magento/CacheInvalidate/Observer/InvalidateVarnishObserver.php

@@ -21,18 +21,6 @@ class InvalidateVarnishObserver implements ObserverInterface
      */
     protected $purgeCache;
 
-    /**
-     * Batch size of the purge request.
-     *
-     * Based on default Varnish 4 http_req_hdr_len size minus a 512 bytes margin for method,
-     * header name, line feeds etc.
-     *
-     * @see http://www.varnish-cache.org/docs/4.0/reference/varnishd.html#http-req-hdr-len
-     *
-     * @var int
-     */
-    private $requestSize = 7680;
-
     /**
      * Invalidation tags resolver
      *
@@ -56,8 +44,6 @@ public function __construct(
      * If Varnish caching is enabled it collects array of tags
      * of incoming object and asks to clean cache.
      *
-     * For scaling reason the purge request is chopped down to fix batch size.
-     *
      * @param \Magento\Framework\Event\Observer $observer
      * @return void
      */
@@ -70,21 +56,10 @@ public function execute(\Magento\Framework\Event\Observer $observer)
         if ($this->config->getType() == \Magento\PageCache\Model\Config::VARNISH && $this->config->isEnabled()) {
             $bareTags = $this->getTagResolver()->getTags($object);
 
-            $tagsBatchSize = 0;
             $tags = [];
-            $pattern = '((^|,)%s(,|$))';
+            $pattern = "((^|,)%s(,|$))";
             foreach ($bareTags as $tag) {
-                $formattedTag = sprintf($pattern, $tag);
-
-                // Send request if batch size is reached and add the implode with pipe to the computation
-                if ($tagsBatchSize + strlen($formattedTag) > $this->requestSize - count($tags) - 1) {
-                    $this->purgeCache->sendPurgeRequest(implode('|', array_unique($tags)));
-                    $tags = [];
-                    $tagsBatchSize = 0;
-                }
-
-                $tagsBatchSize += strlen($formattedTag);
-                $tags[] = $formattedTag;
+                $tags[] = sprintf($pattern, $tag);
             }
             if (!empty($tags)) {
                 $this->purgeCache->sendPurgeRequest(implode('|', array_unique($tags)));