Enhancing the `logs-*-*` index template with a default ingest pipeline that is first doing a pre-flight check if the `message` field might be JSON and then uses the JSON processor to decode the JSON and merge it top-level with the document.

See also this prototype: https://gist.github.com/felixbarny/a9a2f6243153d5508643fd95ac968a88#file-routing-yml-L114-L174

Open questions and things to consider
* How do users opt-out of default JSON parsing? They could override the index template. Maybe that's good enough.
* Similarly to the logs@custom component template, should we call out to a custom index pipeline? Align with naming in Fleet.
x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/template/IndexTemplateRegistry.java
(req, listener) -> client.execute(PutComposableIndexTemplateAction.INSTANCE, req, listener)
            );
        });
    }
    private void addIndexLifecyclePoliciesIfMissing(ClusterState state) {
        Optional<IndexLifecycleMetadata> maybeMeta = Optional.ofNullable(state.metadata().custom(IndexLifecycleMetadata.TYPE));
        for (LifecyclePolicy policy : getPolicyConfigs()) {
            final AtomicBoolean creationCheck = policyCreationsInProgress.computeIfAbsent(
                policy.getName(),
                key -> new AtomicBoolean(false)
            );
            if (creationCheck.compareAndSet(false, true)) {
                final boolean policyNeedsToBeCreated = maybeMeta.flatMap(
                    ilmMeta -> Optional.ofNullable(ilmMeta.getPolicies().get(policy.getName()))
                ).isPresent() == false;
                if (policyNeedsToBeCreated) {
                    logger.debug("adding lifecycle policy [{}] for [{}], because it doesn't exist", policy.getName(), getOrigin());
                    putPolicy(policy, creationCheck);
                } else {
                    logger.trace("not adding lifecycle policy [{}] for [{}], because it already exists", policy.getName(), getOrigin());
                    creationCheck.set(false);
                }
            }
        }
    }
    private void putPolicy(final LifecyclePolicy policy, final AtomicBoolean creationCheck) {
        final Executor executor = threadPool.generic();
        executor.execute(() -> {
            PutLifecycleAction.Request request = new PutLifecycleAction.Request(policy);
            request.masterNodeTimeout(TimeValue.timeValueMinutes(1));