massive update allowing watches to support non-associative array responses

travisghansen · travisghansen · commit 9dfd25f04bda · 2023-10-09T23:37:00.000-04:00
Signed-off-by: Travis Glenn Hansen &lt;travisghansen@yahoo.com&gt;
diff --git a/.gitignore b/.gitignore
@@ -17,6 +17,7 @@ composer.lock
 /php-cs-fixer.phar
 .php_cs.cache
 
+/dev
 
 # always keep .keep files around
 !.keep
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -0,0 +1,9 @@
+# v0.4.0
+
+Released 2023-10-09
+
+- better support for `pcntl` signal handling
+- more control over how requests / responses are handled (allow control of encode/decoding options)
+- support for `ReactPHP` loops
+- small internal library (`Dotty`) useful for interacting with structured data (arrays, stdobject)
+- update composer deps
diff --git a/composer.json b/composer.json
@@ -13,8 +13,8 @@
     "require": {
         "php": ">=7.2.0",
         "ext-json": "*",
-        "symfony/yaml": "^5.0",
-        "flow/jsonpath": "^0.5.0"
+        "symfony/yaml": "^6.3",
+        "softcreatr/jsonpath": "^0.8.3"
     },
     "suggest": {
         "ext-pcntl": "support forking of watches",
diff --git a/src/KubernetesClient/Client.php b/src/KubernetesClient/Client.php
@@ -221,7 +221,7 @@ public function setDefaultRequestOptions($options) {
      * @param $options
      * @return mixed|void
      */
-    protected function getRequestOption($option, $options) {
+    public function getRequestOption($option, $options) {
         $defaults = [
             'encode_flags' => 0,
             'decode_flags' => 0,
diff --git a/src/KubernetesClient/Config.php b/src/KubernetesClient/Config.php
@@ -177,7 +177,7 @@ public static function shutdown()
      * Otherwise try to create a config based off running inside a cluster if corresponding files found.
      *
      * @return Config
-     * @throws \Error If no config can be found at at all the default paths.
+     * @throws \Error If no config can be found at all the default paths.
      */
     public static function LoadFromDefault()
     {
diff --git a/src/KubernetesClient/Dotty/DotAccess.php b/src/KubernetesClient/Dotty/DotAccess.php
@@ -0,0 +1,144 @@
+<?php
+
+namespace KubernetesClient\Dotty;
+
+class DotAccess {
+    private const DELIMITERS = ['.'];
+    protected static function keyToPathArray($path): array
+    {
+        if (is_array($path)) {
+            return $path;
+        }
+
+        if (\strlen($path) === 0) {
+            throw new \Exception('Path cannot be an empty string');
+        }
+
+        $path = \str_replace(self::DELIMITERS, '.', $path);
+        $path = preg_replace('/\[([a-zA-Z0-9_\-]*)\]/', '.${1}', $path);
+
+        return \explode('.', $path);
+    }
+
+    public static function propExists(&$data, $key) {
+        if (is_object($data) && property_exists($data, $key)) {
+            return true;
+        }
+        if (is_array($data) && array_key_exists($key, $data)) {
+            return true;
+        }
+
+        return false;
+    }
+    public static function &propGet(&$data, $key) {
+        if (!self::propExists($data, $key)) {
+            throw new \Exception('property not present');
+        }
+        if (is_object($data)) {
+            return $data->{$key};
+        }
+        if (is_array($data)) {
+            return $data[$key];
+        }
+    }
+
+    public static function propSet(&$data, $key, $value) {
+        if (is_object($data)) {
+            $data->{$key} = $value;
+        }
+        if (is_array($data)) {
+            $data[$key] = $value;
+        }
+    }
+
+    public static function isStructuredData(&$data) {
+        if (is_object($data) || is_array($data)) {
+            return true;
+        }
+        return false;
+    }
+
+    public static function &get(&$currentValue, $key, $default = null) {
+        $hasDefault = \func_num_args() > 2;
+
+        if (is_string($key)) {
+            $keyPath = self::keyToPathArray($key);
+        } else {
+            $keyPath = $key;
+        }
+
+        foreach ($keyPath as $currentKey) {
+            if (!self::isStructuredData($currentValue) || !self::propExists($currentValue, $currentKey)) {
+                if ($hasDefault) {
+                    return $default;
+                }
+
+                throw new \Exception('path not present');
+            }
+
+            $currentValue = &self::propGet($currentValue, $currentKey);
+        }
+
+        return $currentValue === null ? $default : $currentValue;
+    }
+
+    public static function exists(&$currentValue, $key) {
+        if (is_string($key)) {
+            $keyPath = self::keyToPathArray($key);
+        } else {
+            $keyPath = $key;
+        }
+
+        foreach ($keyPath as $currentKey) {
+            if (!self::isStructuredData($currentValue) || !self::propExists($currentValue, $currentKey)) {
+                return false;
+            }
+
+            $currentValue = &self::propGet($currentValue, $currentKey);
+        }
+
+        return true;
+    }
+
+    public static function set(&$data, $key, $value, $options = []) {
+        if (is_string($key)) {
+            $keyPath = self::keyToPathArray($key);
+        } else {
+            $keyPath = $key;
+        }
+
+        $currentValue = &$data;
+
+        $keySize = sizeof($keyPath);
+        for ($i = 0; $i < $keySize; $i++) {
+            $currentKey = $keyPath[$i];
+
+            if ($i == ($keySize - 1)) {
+                self::propSet($currentValue, $currentKey, $value);
+                return;
+            }
+
+            if (!self::isStructuredData($currentValue) && self::propExists($currentValue, $currentKey)) {
+                throw new \Exception("key {$currentKey} already exists but it unstructured content");
+            }
+
+            if (!self::propExists($currentValue, $currentKey)) {
+                // if option to create is enabled create
+                if (self::get($options, 'create_structure', true)) {
+                    $type = self::get($options, 'create_structure_type', 'obj');
+                    if ($type == 'array') {
+                        self::propSet($currentValue, $currentKey, []);
+                    }
+
+                    if ($type == 'obj') {
+                        self::propSet($currentValue, $currentKey, new \stdClass());
+                    }
+                } else {
+                    throw new \Exception('necessary parents do not exist');
+                }
+            }
+
+            $currentValue = &self::propGet($currentValue, $currentKey);
+        }
+    }
+}
diff --git a/src/KubernetesClient/Watch.php b/src/KubernetesClient/Watch.php
@@ -2,6 +2,8 @@
 
 namespace KubernetesClient;
 
+use KubernetesClient\Dotty\DotAccess;
+
 /**
  * Used for the various kubernetes watch endpoints for continuous feed of data
  *
@@ -201,6 +203,7 @@ private function getHandle()
             $handle = @fopen($url, 'r', false, $this->getClient()->getStreamContext());
             if ($handle === false) {
                 $e = error_get_last();
+                var_dump($e);
                 throw new \Exception($e['message'], $e['type']);
             }
             stream_set_timeout($handle, 0, $this->getStreamTimeout());
@@ -408,7 +411,22 @@ private function internal_iterator($cycles = 0)
     {
         $handle = $this->getHandle();
         $i_cycles = 0;
+
+        $associative = $this->getClient()->getRequestOption('decode_associative', []);
+        $decode_flags = $this->getClient()->getRequestOption('decode_flags', []);
+        $decode_response = $this->getClient()->getRequestOption('decode_response', []);
+
+        /**
+         * Mitigation for improper ordering especially during initial load
+         * This acts as a tripwire, once tripped it should never go back to false
+         *
+         * https://github.com/kubernetes/kubernetes/issues/49745
+         */
+        $initial_load_finished = false;
         while (true) {
+            if (function_exists('pcntl_signal_dispatch')) {
+                \pcntl_signal_dispatch();
+            }
             if ($this->getStop()) {
                 $this->internal_stop();
                 return;
@@ -426,7 +444,7 @@ private function internal_iterator($cycles = 0)
 
             //$meta = stream_get_meta_data($handle);
             if (feof($handle)) {
-                if ($this->params['timeoutSeconds'] > 0) {
+                if (key_exists('timeoutSeconds', $this->params) && $this->params['timeoutSeconds'] > 0) {
                     //assume we've reached a successful end of watch
                     return;
                 } else {
@@ -449,6 +467,10 @@ private function internal_iterator($cycles = 0)
                 $this->setLastBytesReadTimestamp(time());
             }
 
+            if (!$initial_load_finished && empty($data)) {
+                $initial_load_finished = true;
+            }
+
             $this->buffer .= $data;
 
             //break immediately if nothing is on the buffer
@@ -462,7 +484,7 @@ private function internal_iterator($cycles = 0)
                 for ($x = 0; $x < ($parts_count - 1); $x++) {
                     if (!empty($parts[$x])) {
                         try {
-                            $response = json_decode($parts[$x], true);
+                            $response = json_decode($parts[$x], $associative, 512, $decode_flags);
                             $code = $this->preProcessResponse($response);
                             if ($code != 0) {
                                 $this->resetHandle();
@@ -471,12 +493,24 @@ private function internal_iterator($cycles = 0)
                                 goto end;
                             }
 
-                            if ($response['object']['metadata']['resourceVersion'] > $this->getResourceVersionLastSuccess()) {
-                                ($this->callback)($response, $this);
+                            if (!$initial_load_finished && DotAccess::get($response, 'type') != "ADDED") {
+                                $initial_load_finished = true;
                             }
 
-                            $this->setResourceVersion($response['object']['metadata']['resourceVersion']);
-                            $this->setResourceVersionLastSuccess($response['object']['metadata']['resourceVersion']);
+                            $rv = DotAccess::get($response, 'object.metadata.resourceVersion');
+
+                            if (!$initial_load_finished || $rv > $this->getResourceVersionLastSuccess()) {
+                                if (!$decode_response) {
+                                    ($this->callback)($parts[$x], $this);
+                                } else {
+                                    ($this->callback)($response, $this);
+                                }
+                            }
+
+                            if ($rv > $this->getResourceVersionLastSuccess()) {
+                                $this->setResourceVersion($rv);
+                                $this->setResourceVersionLastSuccess($rv);
+                            }
 
                             if ($this->getStop()) {
                                 $this->internal_stop();
@@ -509,7 +543,23 @@ private function internal_generator($cycles = 0)
     {
         $handle = $this->getHandle();
         $i_cycles = 0;
+
+        $associative = $this->getClient()->getRequestOption('decode_associative', []);
+        $decode_flags = $this->getClient()->getRequestOption('decode_flags', []);
+        $decode_response = $this->getClient()->getRequestOption('decode_response', []);
+
+        /**
+         * Mitigation for improper ordering especially during initial load
+         * This acts as a tripwire, once tripped it should never go back to false
+         *
+         * https://github.com/kubernetes/kubernetes/issues/49745
+         */
+        $initial_load_finished = false;
         while (true) {
+            if (function_exists('pcntl_signal_dispatch')) {
+                \pcntl_signal_dispatch();
+            }
+
             if ($this->getStop()) {
                 $this->internal_stop();
                 return;
@@ -527,7 +577,7 @@ private function internal_generator($cycles = 0)
 
             //$meta = stream_get_meta_data($handle);
             if (feof($handle)) {
-                if ($this->params['timeoutSeconds'] > 0) {
+                if (key_exists('timeoutSeconds', $this->params) && $this->params['timeoutSeconds'] > 0) {
                     //assume we've reached a successful end of watch
                     return;
                 } else {
@@ -550,6 +600,10 @@ private function internal_generator($cycles = 0)
                 $this->setLastBytesReadTimestamp(time());
             }
 
+            if (!$initial_load_finished && empty($data)) {
+                $initial_load_finished = true;
+            }
+
             $this->buffer .= $data;
 
             //break immediately if nothing is on the buffer
@@ -563,7 +617,7 @@ private function internal_generator($cycles = 0)
                 for ($x = 0; $x < ($parts_count - 1); $x++) {
                     if (!empty($parts[$x])) {
                         try {
-                            $response = json_decode($parts[$x], true);
+                            $response = json_decode($parts[$x], $associative, 512, $decode_flags);
                             $code = $this->preProcessResponse($response);
                             if ($code != 0) {
                                 $this->resetHandle();
@@ -572,13 +626,26 @@ private function internal_generator($cycles = 0)
                                 goto end;
                             }
 
-                            $yield = ($response['object']['metadata']['resourceVersion'] > $this->getResourceVersionLastSuccess());
+                            if (!$initial_load_finished && DotAccess::get($response, 'type') != "ADDED") {
+                                $initial_load_finished = true;
+                            }
+
+                            $rv = DotAccess::get($response, 'object.metadata.resourceVersion');
 
-                            $this->setResourceVersion($response['object']['metadata']['resourceVersion']);
-                            $this->setResourceVersionLastSuccess($response['object']['metadata']['resourceVersion']);
+                            // https://github.com/kubernetes/kubernetes/issues/49745
+                            $yield = (!$initial_load_finished || $rv > $this->getResourceVersionLastSuccess());
+
+                            if ($rv > $this->getResourceVersionLastSuccess()) {
+                                $this->setResourceVersion($rv);
+                                $this->setResourceVersionLastSuccess($rv);
+                            }
 
                             if ($yield) {
-                                yield $response;
+                                if (!$decode_response) {
+                                    yield $parts[$x];
+                                } else {
+                                    yield $response;
+                                }
                             }
 
                             if ($this->getStop()) {
@@ -603,16 +670,16 @@ private function internal_generator($cycles = 0)
 
     private function preProcessResponse($response)
     {
-        if (!is_array($response)) {
+        if (!DotAccess::isStructuredData($response)) {
             return 1;
         }
 
-        if (key_exists('kind', $response) && $response['kind'] == 'Status' && $response['status'] == 'Failure') {
+        if(DotAccess::get($response, 'kind', null) == 'Status' && DotAccess::get($response, 'status', null) == 'Failure') {
             return 1;
         }
 
         // resourceVersion is too old
-        if ($response['type'] == 'ERROR' && $response['object']['code'] == 410) {
+        if (DotAccess::get($response, 'type', null) == 'ERROR' && DotAccess::get($response, 'object.code', null) == 410) {
             return 1;
         }
 

Original file line number	Diff line number	Diff line change
`@@ -177,7 +177,7 @@ public static function shutdown()`
`177`	`177`	`* Otherwise try to create a config based off running inside a cluster if corresponding files found.`
`178`	`178`	`*`
`179`	`179`	`* @return Config`
`180`		`- * @throws \Error If no config can be found at at all the default paths.`
	`180`	`+ * @throws \Error If no config can be found at all the default paths.`
`181`	`181`	`*/`
`182`	`182`	`public static function LoadFromDefault()`
`183`	`183`	`{`