Try a different tactic.

Rather than guessing at the last applied configuration, attempt to read
it from the `kubectl.kubernetes.io/last-applied-configuration`
annotation in the live object state. If this key is not present, no
inputs are populated.

These changes also update the provider to set this field during `Create`
and `Update`.

pkg/provider/provider.go

@@ -58,6 +58,8 @@ import (
 
 // --------------------------------------------------------------------------
 
+const lastAppliedConfigKey = "kubectl.kubernetes.io/last-applied-configuration"
+
 type cancellationContext struct {
 	context context.Context
 	cancel  context.CancelFunc
@@ -492,6 +494,11 @@ func (k *kubeProvider) Create(
 	}
 	newInputs := propMapToUnstructured(newResInputs)
 
+	annotatedInputs, err := withLastAppliedConfig(newInputs)
+	if err != nil {
+		return nil, err
+	}
+
 	config := await.CreateConfig{
 		ProviderConfig: await.ProviderConfig{
 			Context:     k.canceler.context,
@@ -500,7 +507,7 @@ func (k *kubeProvider) Create(
 			ClientSet:   k.clientSet,
 			DedupLogger: logging.NewLogger(k.canceler.context, k.host, urn),
 		},
-		Inputs: newInputs,
+		Inputs: annotatedInputs,
 	}
 
 	initialized, awaitErr := await.Creation(config)
@@ -648,8 +655,11 @@ func (k *kubeProvider) Read(ctx context.Context, req *pulumirpc.ReadRequest) (*p
 		// initialize.
 	}
 
-	// Estimate the inputs for this object.
-	liveInputs := estimateLiveInputs(liveObj)
+	// Attempt to parse the inputs for this object. If parsing was unsuccessful, retain the old inputs.
+	liveInputs := parseLiveInputs(liveObj)
+	if liveInputs == nil {
+		liveInputs = oldInputs
+	}
 
 	// TODO(lblackstone): not sure why this is needed
 	id := FqObjName(liveObj)
@@ -772,6 +782,11 @@ func (k *kubeProvider) Update(
 	}
 	newInputs := propMapToUnstructured(newResInputs)
 
+	annotatedInputs, err := withLastAppliedConfig(newInputs)
+	if err != nil {
+		return nil, err
+	}
+
 	config := await.UpdateConfig{
 		ProviderConfig: await.ProviderConfig{
 			Context:     k.canceler.context,
@@ -781,7 +796,7 @@ func (k *kubeProvider) Update(
 			DedupLogger: logging.NewLogger(k.canceler.context, k.host, urn),
 		},
 		Previous: oldInputs,
-		Inputs:   newInputs,
+		Inputs:   annotatedInputs,
 	}
 	// Apply update.
 	initialized, awaitErr := await.Update(config)
@@ -952,6 +967,26 @@ func propMapToUnstructured(pm resource.PropertyMap) *unstructured.Unstructured {
 	return &unstructured.Unstructured{Object: pm.Mappable()}
 }
 
+func withLastAppliedConfig(config *unstructured.Unstructured) (*unstructured.Unstructured, error) {
+	// Serialize the inputs and add the last-applied-configuration annotation.
+	marshaled, err := config.MarshalJSON()
+	if err != nil {
+		return nil, err
+	}
+
+	// Deep copy the config before returning.
+	config = config.DeepCopy()
+
+	annotations := config.GetAnnotations()
+	if annotations == nil {
+		annotations = make(map[string]string)
+	}
+
+	annotations[lastAppliedConfigKey] = string(marshaled)
+	config.SetAnnotations(annotations)
+	return config, nil
+}
+
 func checkpointObject(inputs, live *unstructured.Unstructured) resource.PropertyMap {
 	object := resource.NewPropertyMapFromMap(live.Object)
 	object["__inputs"] = resource.NewObjectProperty(resource.NewPropertyMapFromMap(inputs.Object))
@@ -1007,24 +1042,23 @@ func canonicalNamespace(ns string) string {
 // deleteResponse causes the resource to be deleted from the state.
 var deleteResponse = &pulumirpc.ReadResponse{Id: "", Properties: nil}
 
-// estimateLiveInputs computes an estimate of the provider inputs that produced the given live object. This is used by
-// Read.
-func estimateLiveInputs(live *unstructured.Unstructured) *unstructured.Unstructured {
-	// Start with a copy of the live object's state.
-	inputs := live.DeepCopy()
-
-	// Remove all known system-populated metadata.
-	inputs.SetUID("")
-	inputs.SetResourceVersion("")
-	inputs.SetGeneration(0)
-	inputs.SetSelfLink("")
-	inputs.SetContinue("")
-	inputs.SetCreationTimestamp(metav1.Time{})
-	inputs.SetDeletionTimestamp(nil)
-	inputs.SetDeletionGracePeriodSeconds(nil)
-
-	// Remove the status field.
-	delete(inputs.Object, "status")
-
-	return inputs
+// parseLiveInputs attempts to parse the provider inputs that produced the given live object out of the object's state.
+// This is used by Read.
+func parseLiveInputs(live *unstructured.Unstructured) *unstructured.Unstructured {
+	// If `kubectl.kubernetes.io/last-applied-configuration` metadata anotation is present, parse it into a real object
+	// and use it as the current set of live inputs. Otherwise, return nil.
+	annotations := live.GetAnnotations()
+	if annotations == nil {
+		return nil
+	}
+	lastAppliedConfig, ok := annotations[lastAppliedConfigKey]
+	if !ok {
+		return nil
+	}
+
+	liveInputs := &unstructured.Unstructured{}
+	if err := liveInputs.UnmarshalJSON([]byte(lastAppliedConfig)); err != nil {
+		return nil
+	}
+	return liveInputs
 }