[FLINK-35493][snapshot] Add historical cleanup for FlinkStateSnapshot CRs

[mateczagany]

Brief change log

• Added InformerEventSource for Flink resources to be able to list linked snapshots during reconciliation
• If FlinkStateSnapshot resources are enabled, clean them up by age/count in SnapshotObserver
• Add unit tests

Verifying this change

• Added unit tests
• Manual testing with checkpoint and snapshot resources, with age and count based cleanup policies

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changes to the CustomResourceDescriptors: no
• Core observer or reconciler logic that is regularly executed: yes

Documentation

• Does this pull request introduce a new feature? yes
• If yes, how is the feature documented? documentation and examples will be provided in PR [FLINK-35267][snapshot] FlinkStateSnapshot documentation and examples #854

[mateczagany]

@anupamaggarwal @gyfora @ferenc-csaky

We need this PR to be able to automatically clean up FlinkStateSnapshot resources created by the operator.

In the current state only periodic snapshots will be cleaned up, I was not sure if we want to also include upgrade snapshots. I would love to hear your inputs on this.

I also created new configs for checkpoints, because I felt like users might want to use different cleanup policies for checkpoints than savepoints.

[gyfora]

Just for clarification, if the snapshot resource is not installed / enabled, will the old cleanup mechanism still work in the legacy mode?

[mateczagany]

I forgot to add a check when setting up informer event sources, the latest commit should take care of that. I have manually checked and the cleanup is still working the same way it has been if CRD is not installed.

[mateczagany]

After talking with @gyfora I have made it so legacy savepoints will be cleaned up even if FlinkStateSnapshot is enabled for the job. This will use the number of completed FlinkStateSnapshot CRs and subtract that from the maxCount. This should help integrating jobs from older operator versions.

I have also changed the logic a bit so we extract metadata.creationTimestamp to determine the age of the snapshot as triggerTimestamp might not always be set (e.g. FlinkStateSnapshot controller did not start).

When this PR gets merged, I will also update #862 to test cleaning up legacy savepoints with FlinkStateSnapshot enabled.

[gyfora]

Not sure if I missed something but seems to me that in the new cleanup logic we can easily end up In situations where all snapshots are deleted which would be good to avoid