Fix shard follow task cleaner under security

[jasontedor]

The shard follow task cleaner executes on behalf of the user to clean up a shard follow task after the follower index has been deleted. Otherwise, these persistent tasks are left laying around, and they fail to execute because the follower index has been deleted. In the face of security, attempts to complete these persistent tasks would fail. This is because these cleanups are executed under the system context (this makes sense, they are happening on behalf of the user after the user has executed an action) but the system role was never granted the permission for persistent task completion. This commit addresses this by adding this cluster privilege to the system role.

Relates #44702
Relates #51971

[elasticmachine]

Pinging @elastic/es-distributed (:Distributed/CCR)

[jasontedor]

This bug report was submitted to me privately, here's the sanitized log message that supports it:

[2020-02-12T19:42:59,579][WARN ][o.e.x.c.a.ShardFollowTaskCleaner] [<active master>] failed to clean up task [FkZOceJrQOqM_dxhKe1C8Q-7]
org.elasticsearch.ElasticsearchSecurityException: action [cluster:admin/persistent/completion] is unauthorized for user [_system]
                at org.elasticsearch.xpack.core.security.support.Exceptions.authorizationError(Exceptions.java:34) ~[?:?]
                at org.elasticsearch.xpack.security.authz.AuthorizationService.denialException(AuthorizationService.java:576) ~[?:?]
                at org.elasticsearch.xpack.security.authz.AuthorizationService.authorizeSystemUser(AuthorizationService.java:380) ~[?:?]
                at org.elasticsearch.xpack.security.authz.AuthorizationService.authorize(AuthorizationService.java:186) ~[?:?]
                at org.elasticsearch.xpack.security.action.filter.SecurityActionFilter.authorizeRequest(SecurityActionFilter.java:170) ~[?:?]
                at org.elasticsearch.xpack.security.action.filter.SecurityActionFilter.lambda$applyInternal$3(SecurityActionFilter.java:156) ~[?:?]
                at org.elasticsearch.action.ActionListener$1.onResponse(ActionListener.java:61) ~[elasticsearch-6.8.7-SNAPSHOT.jar:6.8.7-SNAPSHOT]
                at org.elasticsearch.xpack.security.authc.AuthenticationService$Authenticator.lambda$authenticateAsync$2(AuthenticationService.java:246) ~[?:?]
                at org.elasticsearch.xpack.security.authc.AuthenticationService$Authenticator.lambda$lookForExistingAuthentication$6(AuthenticationService.java:306) ~[?:?]
                at org.elasticsearch.xpack.security.authc.AuthenticationService$Authenticator.lookForExistingAuthentication(AuthenticationService.java:317) ~[?:?]
                at org.elasticsearch.xpack.security.authc.AuthenticationService$Authenticator.authenticateAsync(AuthenticationService.java:244) ~[?:?]
                at org.elasticsearch.xpack.security.authc.AuthenticationService$Authenticator.access$000(AuthenticationService.java:196) ~[?:?]
                at org.elasticsearch.xpack.security.authc.AuthenticationService.authenticate(AuthenticationService.java:139) ~[?:?]
                at org.elasticsearch.xpack.security.action.filter.SecurityActionFilter.applyInternal(SecurityActionFilter.java:153) ~[?:?]
                at org.elasticsearch.xpack.security.action.filter.SecurityActionFilter.lambda$apply$1(SecurityActionFilter.java:90) ~[?:?]
                at org.elasticsearch.xpack.core.security.SecurityContext.executeAsUser(SecurityContext.java:96) ~[?:?]
                at org.elasticsearch.xpack.security.action.filter.SecurityActionFilter.apply(SecurityActionFilter.java:88) ~[?:?]
                at org.elasticsearch.action.support.TransportAction$RequestFilterChain.proceed(TransportAction.java:165) ~[elasticsearch-6.8.7-SNAPSHOT.jar:6.8.7-SNAPSHOT]
                at org.elasticsearch.action.support.TransportAction.execute(TransportAction.java:139) ~[elasticsearch-6.8.7-SNAPSHOT.jar:6.8.7-SNAPSHOT]
                at org.elasticsearch.action.support.TransportAction.execute(TransportAction.java:81) ~[elasticsearch-6.8.7-SNAPSHOT.jar:6.8.7-SNAPSHOT]
                at org.elasticsearch.client.node.NodeClient.executeLocally(NodeClient.java:87) ~[elasticsearch-6.8.7-SNAPSHOT.jar:6.8.7-SNAPSHOT]
                at org.elasticsearch.client.node.NodeClient.doExecute(NodeClient.java:76) ~[elasticsearch-6.8.7-SNAPSHOT.jar:6.8.7-SNAPSHOT]
                at org.elasticsearch.client.support.AbstractClient.execute(AbstractClient.java:403) ~[elasticsearch-6.8.7-SNAPSHOT.jar:6.8.7-SNAPSHOT]
                at org.elasticsearch.xpack.ccr.action.ShardFollowTaskCleaner.lambda$clusterChanged$0(ShardFollowTaskCleaner.java:71) ~[?:?]
                at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [?:1.8.0_212]
                at java.util.concurrent.FutureTask.run(FutureTask.java:266) [?:1.8.0_212]
                at org.elasticsearch.common.util.concurrent.ThreadContext$ContextPreservingRunnable.run(ThreadContext.java:681) [elasticsearch-6.8.7-SNAPSHOT.jar:6.8.7-SNAPSHOT]
                at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [?:1.8.0_212]
                at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [?:1.8.0_212]
                at java.lang.Thread.run(Thread.java:748) [?:1.8.0_212]

[martijnvg]

👍 LGTM

[tlrx]

LGTM 2, thanks for fixing this Jason

[jasontedor]

@tvernum Can you review too, that we’re okay granting this privilege to the system user? I think so, the system is doing something on behalf of the user after an action the user had permissions for has been completed. The other option here is to add this privilege to the manage_ccr cluster role, we can wrap the client with the headers stored in the shard follow task before executing the persistent task completion. This means that users that have the manage_ccr cluster role have access to persistent task completion, which I don’t think carries any risks, but granting this to system is smaller in scope of permissions handed out by this change, it feels system-y anyway. Please let me know your thoughts.

[jasontedor]

@tvernum Since you're unavailable for a few days, I'm going to merge this to keep other work moving. When you're available again, can you please check if you have any concerns with the approach taken here. If so, we can fix it in a follow-up.

[tvernum]

LGTM

[jasontedor]

Thanks for looking @tvernum!