diff --git a/src/test/groovy/StageStatusCacheTests.groovy b/src/test/groovy/StageStatusCacheTests.groovy
new file mode 100644
index 000000000..f26c791a5
--- /dev/null
+++ b/src/test/groovy/StageStatusCacheTests.groovy
@@ -0,0 +1,147 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import org.junit.Before
+import org.junit.Test
+import static org.junit.Assert.assertTrue
+import static org.junit.Assert.assertFalse
+
+class StageStatusCacheTests extends ApmBasePipelineTest {
+
+  @Override
+  @Before
+  void setUp() throws Exception {
+    super.setUp()
+    script = loadScript('vars/stageStatusCache.groovy')
+    env.GIT_BASE_COMMIT = "29480a51"
+    env.STAGE_ID = 'fooo'
+    env.FILE_NAME_BASE64 = 'Zm9vbzI5NDgwYTUxCg'
+    env.BUILD_ID = "10"
+    helper.registerAllowedMethod('googleStorageUploadExt', [Map.class], { "OK" })
+    helper.registerAllowedMethod('isUserTrigger', { false })
+    helper.registerAllowedMethod('base64encode', [Map.class], { env.FILE_NAME_BASE64 })
+  }
+
+  @Test
+  void test() throws Exception {
+    helper.registerAllowedMethod('fileExists', [String.class], { false })
+    def isOK = false
+    script.call(id: env.STAGE_ID){
+      isOK = true
+    }
+    printCallStack()
+    assertTrue(isOK)
+    assertTrue(assertMethodCallContainsPattern('base64encode', "${env.STAGE_ID}${env.GIT_BASE_COMMIT}"))
+    assertTrue(assertMethodCallContainsPattern('cmd', 'Download Stage Status'))
+    assertTrue(assertMethodCallContainsPattern('fileExists', "${env.FILE_NAME_BASE64}"))
+    assertTrue(assertMethodCallContainsPattern('writeFile', "file=${env.FILE_NAME_BASE64}"))
+    assertTrue(assertMethodCallContainsPattern('googleStorageUploadExt', "pattern=${env.FILE_NAME_BASE64}"))
+    assertJobStatusSuccess()
+  }
+
+  @Test
+  void testCache() throws Exception {
+    helper.registerAllowedMethod('fileExists', [String.class], { true })
+    def isOK = true
+    script.call(id: env.STAGE_ID){
+      isOK = false
+    }
+    printCallStack()
+    assertTrue(isOK)
+    assertTrue(assertMethodCallContainsPattern('base64encode', "${env.STAGE_ID}${env.GIT_BASE_COMMIT}"))
+    assertTrue(assertMethodCallContainsPattern('cmd', 'Download Stage Status'))
+    assertTrue(assertMethodCallContainsPattern('fileExists', "${env.FILE_NAME_BASE64}"))
+    assertTrue(assertMethodCallContainsPattern('log', "The stage skiped because it is in the execution cache."))
+    assertJobStatusSuccess()
+  }
+
+  @Test
+  void testNoCheckCacheOnFirstBuild() throws Exception {
+    helper.registerAllowedMethod('fileExists', [String.class], { true })
+    env.BUILD_ID = "1"
+    def isOK = false
+    script.call(id: env.STAGE_ID){
+      isOK = true
+    }
+    printCallStack()
+    assertTrue(isOK)
+    assertTrue(assertMethodCallContainsPattern('base64encode', "${env.STAGE_ID}${env.GIT_BASE_COMMIT}"))
+    assertFalse(assertMethodCallContainsPattern('cmd', 'Download Stage Status'))
+    assertFalse(assertMethodCallContainsPattern('fileExists', "${env.FILE_NAME_BASE64}"))
+    assertTrue(assertMethodCallContainsPattern('writeFile', "file=${env.FILE_NAME_BASE64}"))
+    assertTrue(assertMethodCallContainsPattern('googleStorageUploadExt', "pattern=${env.FILE_NAME_BASE64}"))
+    assertJobStatusSuccess()
+  }
+
+  @Test
+  void testNoCheckCacheOnRunAlways() throws Exception {
+    helper.registerAllowedMethod('fileExists', [String.class], { true })
+    def isOK = false
+    script.call(id: env.STAGE_ID, runAlways: true){
+      isOK = true
+    }
+    printCallStack()
+    assertTrue(isOK)
+    assertTrue(assertMethodCallContainsPattern('base64encode', "${env.STAGE_ID}${env.GIT_BASE_COMMIT}"))
+    assertFalse(assertMethodCallContainsPattern('cmd', 'Download Stage Status'))
+    assertFalse(assertMethodCallContainsPattern('fileExists', "${env.FILE_NAME_BASE64}"))
+    assertTrue(assertMethodCallContainsPattern('writeFile', "file=${env.FILE_NAME_BASE64}"))
+    assertTrue(assertMethodCallContainsPattern('googleStorageUploadExt', "pattern=${env.FILE_NAME_BASE64}"))
+    assertJobStatusSuccess()
+  }
+
+  @Test
+  void testNoCheckCacheOnUserTrigger() throws Exception {
+    helper.registerAllowedMethod('fileExists', [String.class], { true })
+    helper.registerAllowedMethod('isUserTrigger', { true })
+    def isOK = false
+    script.call(id: env.STAGE_ID){
+      isOK = true
+    }
+    printCallStack()
+    assertTrue(isOK)
+    assertTrue(assertMethodCallContainsPattern('base64encode', "${env.STAGE_ID}${env.GIT_BASE_COMMIT}"))
+    assertFalse(assertMethodCallContainsPattern('cmd', 'Download Stage Status'))
+    assertFalse(assertMethodCallContainsPattern('fileExists', "${env.FILE_NAME_BASE64}"))
+    assertTrue(assertMethodCallContainsPattern('writeFile', "file=${env.FILE_NAME_BASE64}"))
+    assertTrue(assertMethodCallContainsPattern('googleStorageUploadExt', "pattern=${env.FILE_NAME_BASE64}"))
+    assertJobStatusSuccess()
+  }
+
+
+  @Test
+  void testParams() throws Exception {
+    helper.registerAllowedMethod('fileExists', [String.class], { false })
+    helper.registerAllowedMethod('isUserTrigger', { true })
+    env.FILE_NAME_BASE64 = 'Zm9vb2Zvb1NIQQo'
+    def isOK = false
+    script.call(id: env.STAGE_ID,
+      bucket: 'bucketFoo',
+      credentialsId: 'fooCredentials',
+      sha: 'fooSHA'){
+      isOK = true
+    }
+    printCallStack()
+    assertTrue(isOK)
+    assertTrue(assertMethodCallContainsPattern('base64encode', "fooofooSHA"))
+    assertTrue(assertMethodCallContainsPattern('writeFile', "file=${env.FILE_NAME_BASE64}"))
+    assertTrue(assertMethodCallContainsPattern('googleStorageUploadExt', "pattern=${env.FILE_NAME_BASE64}"))
+    assertTrue(assertMethodCallContainsPattern('googleStorageUploadExt', "bucket=gs://bucketFoo/ci/cache/"))
+    assertTrue(assertMethodCallContainsPattern('googleStorageUploadExt', "credentialsId=fooCredentials"))
+    assertJobStatusSuccess()
+  }
+}
diff --git a/vars/README.md b/vars/README.md
index fbcef0838..d2d059499 100644
--- a/vars/README.md
+++ b/vars/README.md
@@ -2350,6 +2350,42 @@ setupAPMGitEmail(global: true)
   stackVersions.edge(snapshot: true) // '8.0.0-SNAPSHOT'
 ```
 
+## stageStatusCache
+Stage status cache allow to save and restore the status of a stage for a particular commit.
+This allow to skip stages when we know that we executed that stage for that commit.
+To do that the step save a file based on `stageSHA|base64` on a GCP bucket,
+this status is checked and execute the body if there is not stage status file
+for the stage and the commit we are building.
+User triggered builds will execute all stages always.
+If the stage success the status is save in a file.
+It uses `GIT_BASE_COMMIT` as a commit SHA, because is a known real commit SHA,
+because of that merges with target branch will skip stages on changes only on target branch.
+
+```
+pipeline {
+  agent any
+  stages {
+    stage('myStage') {
+      steps {
+        deleteDir()
+        stageStatusCache(id: 'myStage',
+          bucket: 'myBucket',
+          credentialsId: 'my-credentials',
+          sha: getGitCommitSha()
+        ){
+          echo "My code"
+        }
+      }
+    }
+  }
+}
+```
+
+* *id:* Unique stage name. Mandatory
+* *bucket:* bucket name. Default 'beats-ci-temp'
+* *credentialsId:* credentials file, with the GCP credentials JSON file. Default  'beats-ci-gcs-plugin-file-credentials'
+* *sha:* Commit SHA used for the stage ID. Default: env.GIT_BASE_COMMIT
+
 ## stashV2
 Stash the current location, for such it compresses the current path and
 upload it to Google Storage.
diff --git a/vars/stageStatusCache.groovy b/vars/stageStatusCache.groovy
new file mode 100644
index 000000000..284b572aa
--- /dev/null
+++ b/vars/stageStatusCache.groovy
@@ -0,0 +1,72 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+/**
+  Execute the body if there is not stage status file for the stage and the commit we are building.
+  User triggered builds will execute all stages always.
+  If the stage success the status is save in a file.
+*/
+def call(Map args, Closure body){
+  if(!args.containsKey('bucket')){
+    args['bucket'] = 'beats-ci-temp'
+  }
+  if(!args.containsKey('credentialsId')){
+    args['credentialsId'] = 'beats-ci-gcs-plugin-file-credentials'
+  }
+  if(!args.containsKey('sha')){
+    args['sha'] = "${env.GIT_BASE_COMMIT}"
+  }
+  if(env.BUILD_ID == "1" || isUserTrigger() || args.get('runAlways', false) || readStageStatus(args) == false){
+    body()
+    saveStageStatus(args)
+  } else {
+    log(level: 'INFO', text: "The stage skiped because it is in the execution cache.")
+  }
+}
+
+/**
+  Save the status file of the stage.
+*/
+def saveStageStatus(Map args){
+  def statusFileName = stageStatusId(args)
+  writeFile(file: statusFileName, text: "OK")
+  googleStorageUploadExt(bucket: "gs://${args.bucket}/ci/cache/",
+    credentialsId: "${args.credentialsId}",
+    pattern: "${statusFileName}",
+    sharedPublicly: true)
+}
+
+/**
+  Read the status file of the stage if it exists.
+*/
+def readStageStatus(Map args){
+  def statusFileName = stageStatusId(args)
+  try {
+    cmd(label: 'Download Stage Status',
+      script: "curl -sSf -O https://storage.googleapis.com/${args.bucket}/ci/cache/${statusFileName}",
+      returnStatus: true)
+  } finally {
+    return fileExists("${statusFileName}")
+  }
+}
+
+/**
+  generate an unique ID for the stage and commit.
+*/
+def stageStatusId(Map args){
+  return base64encode(text: "${args.id}${args.sha}", encoding: "UTF-8", padding: false)
+}
diff --git a/vars/stageStatusCache.txt b/vars/stageStatusCache.txt
new file mode 100644
index 000000000..a1718eb02
--- /dev/null
+++ b/vars/stageStatusCache.txt
@@ -0,0 +1,34 @@
+Stage status cache allow to save and restore the status of a stage for a particular commit.
+This allow to skip stages when we know that we executed that stage for that commit.
+To do that the step save a file based on `stageSHA|base64` on a GCP bucket,
+this status is checked and execute the body if there is not stage status file
+for the stage and the commit we are building.
+User triggered builds will execute all stages always.
+If the stage success the status is save in a file.
+It uses `GIT_BASE_COMMIT` as a commit SHA, because is a known real commit SHA,
+because of that merges with target branch will skip stages on changes only on target branch.
+
+```
+pipeline {
+  agent any
+  stages {
+    stage('myStage') {
+      steps {
+        deleteDir()
+        stageStatusCache(id: 'myStage',
+          bucket: 'myBucket',
+          credentialsId: 'my-credentials',
+          sha: getGitCommitSha()
+        ){
+          echo "My code"
+        }
+      }
+    }
+  }
+}
+```
+
+* *id:* Unique stage name. Mandatory
+* *bucket:* bucket name. Default 'beats-ci-temp'
+* *credentialsId:* credentials file, with the GCP credentials JSON file. Default  'beats-ci-gcs-plugin-file-credentials'
+* *sha:* Commit SHA used for the stage ID. Default: env.GIT_BASE_COMMIT