From 395bfa2f5824cf6e55ce814b7a78bc9ea130fec5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kristo=20Kuusk=C3=BCll?= Date: Tue, 30 May 2023 15:21:58 +0300 Subject: [PATCH] Improved the query parsing for table access statistics (#31) * Improvements for Table Access Statistics query parsing. --- CHANGELOG.md | 21 +++ README.md | 2 - docs/index.md | 72 +++++++- gradle.properties | 2 +- .../EntryPointsAutoConfiguration.java | 29 +++ ...ableAccessStatisticsBeanPostProcessor.java | 22 ++- .../TableAccessStatisticsIntTest.java | 64 ++++++- .../TestTasQueryParsingInterceptor.java | 21 +++ .../entrypoints/test/TestApplication.java | 6 + tw-entrypoints/build-idea-fix.gradle.kts | 6 + tw-entrypoints/build.gradle | 3 +- .../entrypoints/EntryPointsProperties.java | 9 +- .../DefaultTasParsedQueryRegistry.java | 19 ++ .../DefaultTasQueryParsingInterceptor.java | 15 ++ .../DefaultTasQueryParsingListener.java | 45 +++++ .../tableaccessstatistics/ParsedQuery.java | 32 ++++ .../tableaccessstatistics/SqlParser.java | 84 +++++++++ .../tableaccessstatistics/SqlParserUtils.java | 37 ---- .../TableAccessStatisticsSpyqlListener.java | 170 +++++++++++------- .../TasParsedQueryRegistry.java | 11 ++ .../TasQueryParsingInterceptor.java | 36 ++++ .../TasQueryParsingListener.java | 10 ++ .../SqlParserUtilsTest.java | 23 ++- 23 files changed, 616 insertions(+), 123 deletions(-) create mode 100644 tw-entrypoints-starter/src/test/java/com/transferwise/common/entrypoints/tableaccessstatistics/TestTasQueryParsingInterceptor.java create mode 100644 tw-entrypoints/build-idea-fix.gradle.kts rename {tw-entrypoints-starter => tw-entrypoints}/src/main/java/com/transferwise/common/entrypoints/EntryPointsProperties.java (68%) create mode 100644 tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/DefaultTasParsedQueryRegistry.java create mode 100644 tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/DefaultTasQueryParsingInterceptor.java create mode 100644 tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/DefaultTasQueryParsingListener.java create mode 100644 tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/ParsedQuery.java create mode 100644 tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/SqlParser.java delete mode 100644 tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/SqlParserUtils.java create mode 100644 tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/TasParsedQueryRegistry.java create mode 100644 tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/TasQueryParsingInterceptor.java create mode 100644 tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/TasQueryParsingListener.java diff --git a/CHANGELOG.md b/CHANGELOG.md index 1d1adc8..ed8776d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,27 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [2.11.0] - 2023-05-18 + +### Changed + +* Implemented a timeout and interruption for TAS SQL parsing. + Complex queries in one of our services created long duration heavy CPU burn. + +* Query parsing will use `JSQLParser` complex parsing immediately. + Before, our implementation was using simple parsing. And `JSQLParser` implementation tried by default `simple` first + and then `complex`, if `simple` failed. + Performance tests showed `simple` parsing for simple queries, is not noticeably faster for simple queries. + +* Created a mechanism for a service to provide parsed query information itself and thus skip the query parsing. + It can be used for complex queries where parsing is slow or for queries which jsqlparser can not handle. + The mechanism can be used via `TableAccessStatisticsParsedQueryRegistry` interface. + +* Added more flexibility around query parsing via `TasQueryParsingListener` and `TasQueryParsingInterceptor`. + +* Supporting parsing queries with `on conflict (...)` clause with multiple parameters. + We can remove our own solution, when next `JSQLParser` version would support it. + ## [2.10.0] - 2023-05-09 ### Added diff --git a/README.md b/README.md index 949df04..11f98e3 100644 --- a/README.md +++ b/README.md @@ -10,5 +10,3 @@ Provides various metrics for service's databases usage. Reference the [documentation](docs/index.md) of the library for more information about adoption and features. - - diff --git a/docs/index.md b/docs/index.md index f602ee2..eba0e56 100644 --- a/docs/index.md +++ b/docs/index.md @@ -1,6 +1,7 @@ # Tw-EntryPoints documentation ## Table of Contents + * [Introduction](#intro) * [Setup](#setup) * [Integration tests](#integration-tests) @@ -18,13 +19,11 @@ Is built on top of Integrates nicely with Transferwise Entrypoints system. - Example dashboards: - [EntryPoints Database Access V3](https://dashboards.tw.ee/d/f6l4lrUWz/entrypoints-database-access-v3?orgId=1) - [EntryPoints Table Access V2](https://dashboards.tw.ee/d/dyp0u9UZz/entrypoints-table-access-v2?orgId=1) - ## Setup Assuming, your data sources are using [HikariCP](https://github.com/brettwooldridge/HikariCP), only thing you need to do, is to add a dependency. @@ -59,6 +58,75 @@ You would need to add a dependency for this as well: testImplementation "com.transferwise.common:tw-entrypoints" ``` +## Table access statistics and `JSqlParser` library + +We are using (JSqlParser)[https://github.com/JSQLParser/JSqlParser] library to parse table names from queries. + +The library is pretty good, but some services have few queries, it can not parse. Also, sometimes the parsing can take so long, +that it will create latency spikes or cpu burns. + +In those case, you can override/control the parsing via `TasQueryParsingInterceptor` and `TasParsedQueryRegistry`. + +Example for `TasQueryParsingInterceptor`. + + +```java +public class MyTasQueryParsingInterceptor extends DefaultTasQueryParsingInterceptor { + + @Override + public InterceptResult intercept(String sql) { + if (StringUtils.startsWithIgnoreCase(sql, "SET fancy_variable TO")) { + return InterceptResult.doSkip(); + } + + else if (sql.equals(knownUnParseableSql)){ + return InterceptResult.returnParsedQuery(new ParsedQuery() + .addOperation("insert",new SqlOperation() + // Main table should always be first, as we register "first-table" metrics by that. + .addTable("transfer") + .addTable("payout"))); + } + + return super.intercept(sql); + } +} +``` + + +Example for `TasParsedQueryRegistry`. + + +```java +@Autowired +private TasParsedQueryRegistry registry; + +public void registerBadSqls(){ + registry.register(knownUnParseableSql,new ParsedQuery() + .addOperation("insert",new SqlOperation() + .addTable("transfer") + .addTable("payout"))); +} +``` + + +In case where failed parsing will create too much logs noise, you have an option to override `TasQueryParsingListener`. + +For example: + +```java +public class MyTasQueryParsingListener extends DefaultTasQueryParsingListener { + + @Override + public void parsingFailed(String sql, Duration timeTaken, Throwable t) { + if (sql.equals(knownProblematicQuery)) { + // ignore + } else { + super.parsingFailed(sql, timeTaken, t); + } + } +} +``` + ## License Copyright 2021 TransferWise Ltd. diff --git a/gradle.properties b/gradle.properties index 4a35443..5f589ef 100644 --- a/gradle.properties +++ b/gradle.properties @@ -1 +1 @@ -version=2.10.0 +version=2.11.0 diff --git a/tw-entrypoints-starter/src/main/java/com/transferwise/common/entrypoints/EntryPointsAutoConfiguration.java b/tw-entrypoints-starter/src/main/java/com/transferwise/common/entrypoints/EntryPointsAutoConfiguration.java index 82e0806..e2d4a37 100644 --- a/tw-entrypoints-starter/src/main/java/com/transferwise/common/entrypoints/EntryPointsAutoConfiguration.java +++ b/tw-entrypoints-starter/src/main/java/com/transferwise/common/entrypoints/EntryPointsAutoConfiguration.java @@ -9,12 +9,19 @@ import com.transferwise.common.entrypoints.databaseaccessstatistics.DatabaseAccessStatisticsBeanPostProcessor; import com.transferwise.common.entrypoints.databaseaccessstatistics.DatabaseAccessStatisticsEntryPointInterceptor; import com.transferwise.common.entrypoints.executionstatistics.ExecutionStatisticsEntryPointInterceptor; +import com.transferwise.common.entrypoints.tableaccessstatistics.DefaultTasParsedQueryRegistry; +import com.transferwise.common.entrypoints.tableaccessstatistics.DefaultTasQueryParsingInterceptor; +import com.transferwise.common.entrypoints.tableaccessstatistics.DefaultTasQueryParsingListener; import com.transferwise.common.entrypoints.tableaccessstatistics.TableAccessStatisticsBeanPostProcessor; +import com.transferwise.common.entrypoints.tableaccessstatistics.TasParsedQueryRegistry; +import com.transferwise.common.entrypoints.tableaccessstatistics.TasQueryParsingInterceptor; +import com.transferwise.common.entrypoints.tableaccessstatistics.TasQueryParsingListener; import com.transferwise.common.entrypoints.transactionstatistics.TransactionStatisticsBeanPostProcessor; import io.micrometer.core.instrument.MeterRegistry; import org.springframework.beans.factory.BeanFactory; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -22,6 +29,7 @@ public class EntryPointsAutoConfiguration { @Bean + @ConfigurationProperties(value = "tw-entrypoints", ignoreUnknownFields = false) public EntryPointsProperties twEntryPointsProperties() { return new EntryPointsProperties(); } @@ -49,6 +57,27 @@ public TableAccessStatisticsBeanPostProcessor twEntryPointsTableAccessStatistics return new TableAccessStatisticsBeanPostProcessor(beanFactory); } + @Bean + @ConditionalOnProperty(name = "tw-entrypoints.tas.enabled", havingValue = "true", matchIfMissing = true) + @ConditionalOnMissingBean(TasParsedQueryRegistry.class) + public DefaultTasParsedQueryRegistry twEntryPointsTableAccessStatisticsParsedQueryRegistry() { + return new DefaultTasParsedQueryRegistry(); + } + + @Bean + @ConditionalOnProperty(name = "tw-entrypoints.tas.enabled", havingValue = "true", matchIfMissing = true) + @ConditionalOnMissingBean(TasQueryParsingInterceptor.class) + public DefaultTasQueryParsingInterceptor twEntryPointsTableAccessStatisticsQueryParsingInterceptor() { + return new DefaultTasQueryParsingInterceptor(); + } + + @Bean + @ConditionalOnProperty(name = "tw-entrypoints.tas.enabled", havingValue = "true", matchIfMissing = true) + @ConditionalOnMissingBean(TasQueryParsingListener.class) + public DefaultTasQueryParsingListener twEntryPointsTableAccessStatisticsQueryParsingListener(EntryPointsProperties entryPointsProperties) { + return new DefaultTasQueryParsingListener(entryPointsProperties); + } + @Bean @ConditionalOnProperty(name = "tw-entrypoints.ts.enabled", havingValue = "true", matchIfMissing = true) @ConditionalOnMissingBean diff --git a/tw-entrypoints-starter/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/TableAccessStatisticsBeanPostProcessor.java b/tw-entrypoints-starter/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/TableAccessStatisticsBeanPostProcessor.java index 7496804..30598d5 100644 --- a/tw-entrypoints-starter/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/TableAccessStatisticsBeanPostProcessor.java +++ b/tw-entrypoints-starter/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/TableAccessStatisticsBeanPostProcessor.java @@ -4,16 +4,12 @@ import com.transferwise.common.baseutils.concurrency.ThreadNamingExecutorServiceWrapper; import com.transferwise.common.baseutils.meters.cache.IMeterCache; import com.transferwise.common.entrypoints.BaseEntryPointsBeanProcessor; +import com.transferwise.common.entrypoints.EntryPointsProperties; import com.transferwise.common.spyql.SpyqlDataSource; -import java.util.concurrent.ExecutorService; import org.springframework.beans.factory.BeanFactory; -import org.springframework.beans.factory.annotation.Value; public class TableAccessStatisticsBeanPostProcessor extends BaseEntryPointsBeanProcessor { - @Value("${tw-entrypoints.tas.sql-parser.cache-size-mib:50}") - private int sqlParserCacheSizeMib; - private final BeanFactory beanFactory; public TableAccessStatisticsBeanPostProcessor(BeanFactory beanFactory) { @@ -23,17 +19,25 @@ public TableAccessStatisticsBeanPostProcessor(BeanFactory beanFactory) { @Override protected void instrument(SpyqlDataSource spyqlDataSource, String databaseName) { boolean isAlreadyAttached = spyqlDataSource.getDataSourceListeners().stream().anyMatch( - l -> l instanceof TableAccessStatisticsSpyqlListener); + TableAccessStatisticsSpyqlListener.class::isInstance); if (isAlreadyAttached) { return; } - IMeterCache meterCache = beanFactory.getBean(IMeterCache.class); - ExecutorService executorService = new ThreadNamingExecutorServiceWrapper("eptas", beanFactory + var entryPointsProperties = beanFactory.getBean(EntryPointsProperties.class); + var meterCache = beanFactory.getBean(IMeterCache.class); + var executorService = new ThreadNamingExecutorServiceWrapper("eptas", beanFactory .getBean(IExecutorServicesProvider.class).getGlobalExecutorService()); + var tableAccessStatisticsParsedQueryRegistry = beanFactory.getBean( + TasParsedQueryRegistry.class); + + var tasQueryParsingInterceptor = beanFactory.getBean(TasQueryParsingInterceptor.class); + var tasQueryParsingListener = beanFactory.getBean(TasQueryParsingListener.class); + spyqlDataSource.addListener( - new TableAccessStatisticsSpyqlListener(meterCache, executorService, databaseName, sqlParserCacheSizeMib)); + new TableAccessStatisticsSpyqlListener(meterCache, executorService, tableAccessStatisticsParsedQueryRegistry, databaseName, + entryPointsProperties, tasQueryParsingListener, tasQueryParsingInterceptor)); } } diff --git a/tw-entrypoints-starter/src/test/java/com/transferwise/common/entrypoints/tableaccessstatistics/TableAccessStatisticsIntTest.java b/tw-entrypoints-starter/src/test/java/com/transferwise/common/entrypoints/tableaccessstatistics/TableAccessStatisticsIntTest.java index cf5217b..3c6d551 100644 --- a/tw-entrypoints-starter/src/test/java/com/transferwise/common/entrypoints/tableaccessstatistics/TableAccessStatisticsIntTest.java +++ b/tw-entrypoints-starter/src/test/java/com/transferwise/common/entrypoints/tableaccessstatistics/TableAccessStatisticsIntTest.java @@ -4,6 +4,7 @@ import com.transferwise.common.baseutils.ExceptionUtils; import com.transferwise.common.context.TwContext; +import com.transferwise.common.entrypoints.tableaccessstatistics.ParsedQuery.SqlOperation; import com.transferwise.common.entrypoints.test.BaseIntTest; import com.transferwise.common.spyql.SpyqlDataSource; import io.micrometer.core.instrument.Counter; @@ -19,11 +20,17 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.jdbc.core.JdbcTemplate; -public class TableAccessStatisticsIntTest extends BaseIntTest { +class TableAccessStatisticsIntTest extends BaseIntTest { + + @Autowired + private TestTasQueryParsingInterceptor testTasQueryParsingInterceptor; @Autowired private DataSource dataSource; + @Autowired + private DefaultTasParsedQueryRegistry tableAccessStatisticsParsedQueryRegistry; + private JdbcTemplate jdbcTemplate; @BeforeEach @@ -31,10 +38,11 @@ public void setup() { super.setup(); jdbcTemplate = new JdbcTemplate(dataSource); invalidateParserCache(); + testTasQueryParsingInterceptor.setParsedQuery(null); } @Test - public void selectToNotExistingTableGetsCorrectlyRegistered() { + void selectToNotExistingTableGetsCorrectlyRegistered() { TwContext.current().createSubContext().asEntryPoint("Test", "myEntryPoint").execute(() -> { try { jdbcTemplate.queryForObject("select id from not_existing_table", Long.class); @@ -71,6 +79,58 @@ public void selectToNotExistingTableGetsCorrectlyRegistered() { assertThat(firstTableAccessMeter.mean(TimeUnit.NANOSECONDS)).isGreaterThan(0); } + @Test + void parsedQueryRegistryCanOverrideParsing() { + String sql = "select id from not_existing_table limit 1234"; + + tableAccessStatisticsParsedQueryRegistry.register(sql, new ParsedQuery().addOperation("upsert", new SqlOperation().addTable("my_custom_table"))); + + TwContext.current().createSubContext().asEntryPoint("Test", "myEntryPoint").execute(() -> { + try { + jdbcTemplate.queryForObject("select id from not_existing_table limit 1234", Long.class); + } catch (Exception ignored) { + //ignored + } + }); + + List meters = getTableAccessMeters(); + + assertThat(meters.size()).isEqualTo(1); + var counter = (Counter) meters.get(0); + + assertThat(counter.getId().getTag("operation")).isEqualTo("upsert"); + assertThat(counter.getId().getTag("table")).isEqualTo("my_custom_table"); + + var firstTableAccessMeter = (Timer) getMeter("EntryPoints_Tas_FirstTableAccess"); + + assertThat(firstTableAccessMeter.getId().getTag("operation")).isEqualTo("upsert"); + assertThat(firstTableAccessMeter.getId().getTag("table")).isEqualTo("my_custom_table"); + } + + @Test + void interceptorCanProvideItsOwnParsedQuery() { + testTasQueryParsingInterceptor.setParsedQuery(new ParsedQuery().addOperation("insert", + new SqlOperation().addTable("my_custom_table123"))); + + String sql = "select id from not_existing_table limit 1234"; + + TwContext.current().createSubContext().asEntryPoint("Test", "myEntryPoint").execute(() -> { + try { + jdbcTemplate.queryForObject(sql, Long.class); + } catch (Exception ignored) { + //ignored + } + }); + + List meters = getTableAccessMeters(); + + assertThat(meters.size()).isEqualTo(1); + var counter = (Counter) meters.get(0); + + assertThat(counter.getId().getTag("operation")).isEqualTo("insert"); + assertThat(counter.getId().getTag("table")).isEqualTo("my_custom_table123"); + } + @Test public void workingUpdateSqlGetCorrectlyRegistered() { TwContext.current().createSubContext().asEntryPoint("Test", "myEntryPoint").execute(() -> { diff --git a/tw-entrypoints-starter/src/test/java/com/transferwise/common/entrypoints/tableaccessstatistics/TestTasQueryParsingInterceptor.java b/tw-entrypoints-starter/src/test/java/com/transferwise/common/entrypoints/tableaccessstatistics/TestTasQueryParsingInterceptor.java new file mode 100644 index 0000000..0d1b43a --- /dev/null +++ b/tw-entrypoints-starter/src/test/java/com/transferwise/common/entrypoints/tableaccessstatistics/TestTasQueryParsingInterceptor.java @@ -0,0 +1,21 @@ +package com.transferwise.common.entrypoints.tableaccessstatistics; + +import lombok.Getter; +import lombok.Setter; +import org.springframework.stereotype.Component; + +@Component +public class TestTasQueryParsingInterceptor extends DefaultTasQueryParsingInterceptor { + + @Getter + @Setter + private ParsedQuery parsedQuery; + + @Override + public InterceptResult intercept(String sql) { + if (parsedQuery != null) { + return InterceptResult.returnParsedQuery(parsedQuery); + } + return super.intercept(sql); + } +} diff --git a/tw-entrypoints-starter/src/test/java/com/transferwise/common/entrypoints/test/TestApplication.java b/tw-entrypoints-starter/src/test/java/com/transferwise/common/entrypoints/test/TestApplication.java index 780016d..937dbf1 100644 --- a/tw-entrypoints-starter/src/test/java/com/transferwise/common/entrypoints/test/TestApplication.java +++ b/tw-entrypoints-starter/src/test/java/com/transferwise/common/entrypoints/test/TestApplication.java @@ -1,6 +1,7 @@ package com.transferwise.common.entrypoints.test; import com.transferwise.common.baseutils.transactionsmanagement.TransactionsHelper; +import com.transferwise.common.entrypoints.tableaccessstatistics.TestTasQueryParsingInterceptor; import com.transferwise.common.spyql.SpyqlDataSource; import com.zaxxer.hikari.HikariConfig; import com.zaxxer.hikari.HikariDataSource; @@ -37,4 +38,9 @@ public DataSource dataSource() { public TransactionsHelper transactionsHelper() { return new TransactionsHelper(); } + + @Bean + public TestTasQueryParsingInterceptor testTasQueryParsingInterceptor() { + return new TestTasQueryParsingInterceptor(); + } } diff --git a/tw-entrypoints/build-idea-fix.gradle.kts b/tw-entrypoints/build-idea-fix.gradle.kts new file mode 100644 index 0000000..747e040 --- /dev/null +++ b/tw-entrypoints/build-idea-fix.gradle.kts @@ -0,0 +1,6 @@ +gradle.taskGraph.whenReady { + val task = this.allTasks.find { it.name.endsWith(".main()") } as? JavaExec // or whatever other method your Main class runs + task?.let { + it.setExecutable(it.javaLauncher.get().executablePath.asFile.absolutePath) + } +} diff --git a/tw-entrypoints/build.gradle b/tw-entrypoints/build.gradle index ce9d53d..4fa8876 100644 --- a/tw-entrypoints/build.gradle +++ b/tw-entrypoints/build.gradle @@ -20,5 +20,6 @@ dependencies { implementation libraries.twGracefulShutdownInterfaces testImplementation libraries.junitJupiter - } + +apply from: "build-idea-fix.gradle.kts" \ No newline at end of file diff --git a/tw-entrypoints-starter/src/main/java/com/transferwise/common/entrypoints/EntryPointsProperties.java b/tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/EntryPointsProperties.java similarity index 68% rename from tw-entrypoints-starter/src/main/java/com/transferwise/common/entrypoints/EntryPointsProperties.java rename to tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/EntryPointsProperties.java index ca4d296..f0d5617 100644 --- a/tw-entrypoints-starter/src/main/java/com/transferwise/common/entrypoints/EntryPointsProperties.java +++ b/tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/EntryPointsProperties.java @@ -1,10 +1,9 @@ package com.transferwise.common.entrypoints; +import java.time.Duration; import lombok.Data; -import org.springframework.boot.context.properties.ConfigurationProperties; @Data -@ConfigurationProperties(value = "tw-entrypoints", ignoreUnknownFields = false) public class EntryPointsProperties { private Das das = new Das(); @@ -34,6 +33,12 @@ public static class Tas { public static class SqlParser { private int cacheSizeMib = 50; + private Duration timeout = Duration.ofSeconds(5); + /** + * If parsing takes longer than that, the service owner would want to know about it. + */ + private Duration parseDurationWarnThreshold = Duration.ofSeconds(1); + private boolean warnAboutFailedParses = true; } } diff --git a/tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/DefaultTasParsedQueryRegistry.java b/tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/DefaultTasParsedQueryRegistry.java new file mode 100644 index 0000000..bfc0405 --- /dev/null +++ b/tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/DefaultTasParsedQueryRegistry.java @@ -0,0 +1,19 @@ +package com.transferwise.common.entrypoints.tableaccessstatistics; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +public class DefaultTasParsedQueryRegistry implements TasParsedQueryRegistry { + + private final Map sqlParseResults = new ConcurrentHashMap<>(); + + @Override + public void register(String sql, ParsedQuery parsedQuery) { + sqlParseResults.put(sql, parsedQuery); + } + + @Override + public ParsedQuery get(String sql) { + return sqlParseResults.get(sql); + } +} diff --git a/tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/DefaultTasQueryParsingInterceptor.java b/tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/DefaultTasQueryParsingInterceptor.java new file mode 100644 index 0000000..8ba7d23 --- /dev/null +++ b/tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/DefaultTasQueryParsingInterceptor.java @@ -0,0 +1,15 @@ +package com.transferwise.common.entrypoints.tableaccessstatistics; + +import org.apache.commons.lang3.StringUtils; + +public class DefaultTasQueryParsingInterceptor implements TasQueryParsingInterceptor { + + @Override + public InterceptResult intercept(String sql) { + if (StringUtils.startsWithIgnoreCase(sql, "SET statement_timeout TO")) { + return InterceptResult.doSkip(); + } + + return InterceptResult.doContinue(); + } +} diff --git a/tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/DefaultTasQueryParsingListener.java b/tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/DefaultTasQueryParsingListener.java new file mode 100644 index 0000000..be391f7 --- /dev/null +++ b/tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/DefaultTasQueryParsingListener.java @@ -0,0 +1,45 @@ +package com.transferwise.common.entrypoints.tableaccessstatistics; + +import com.transferwise.common.entrypoints.EntryPointsProperties; +import java.time.Duration; +import lombok.extern.slf4j.Slf4j; + +@Slf4j +public class DefaultTasQueryParsingListener implements TasQueryParsingListener { + + public static final String FAILED_PARSING_SUGGESTION = " You may want to use `TableAccessStatisticsParsedQueryRegistry` " + + "or provide `DefaultTasQueryParsingInterceptor` implementation to provide the " + + "query parse result manually."; + public static final String FAILED_PARSING_LOG_MESSAGE = "Parsing statement '{}' failed." + FAILED_PARSING_SUGGESTION; + public static final String SLOW_PARSING_LOG_MESSAGE = "Statement '{}' parsing took {} ms." + FAILED_PARSING_LOG_MESSAGE; + + + private EntryPointsProperties entryPointsProperties; + + public DefaultTasQueryParsingListener(EntryPointsProperties entryPointsProperties) { + this.entryPointsProperties = entryPointsProperties; + } + + @Override + public void parsingDone(String sql, ParsedQuery parsedQuery, Duration timeTaken) { + handleParsingDuration(sql, timeTaken); + } + + + @Override + public void parsingFailed(String sql, Duration timeTaken, Throwable t) { + if (entryPointsProperties.getTas().getSqlParser().isWarnAboutFailedParses()) { + log.warn(FAILED_PARSING_LOG_MESSAGE, sql, t); + } else { + log.debug(FAILED_PARSING_LOG_MESSAGE, sql, t); + } + + handleParsingDuration(sql, timeTaken); + } + + protected void handleParsingDuration(String sql, Duration timeTaken) { + if (timeTaken.toMillis() > entryPointsProperties.getTas().getSqlParser().getParseDurationWarnThreshold().toMillis()) { + log.warn(SLOW_PARSING_LOG_MESSAGE, sql, timeTaken.toMillis()); + } + } +} diff --git a/tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/ParsedQuery.java b/tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/ParsedQuery.java new file mode 100644 index 0000000..b290bdb --- /dev/null +++ b/tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/ParsedQuery.java @@ -0,0 +1,32 @@ +package com.transferwise.common.entrypoints.tableaccessstatistics; + +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; +import lombok.Data; +import lombok.experimental.Accessors; + +@Data +@Accessors(chain = true) +public class ParsedQuery { + + private Map operations = new HashMap<>(); + + ParsedQuery addOperation(String operationName, SqlOperation operation) { + operations.put(operationName, operation); + return this; + } + + @Data + @Accessors(chain = true) + static class SqlOperation { + + private Set tableNames = new HashSet<>(); + + public SqlOperation addTable(String table) { + tableNames.add(table); + return this; + } + } +} diff --git a/tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/SqlParser.java b/tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/SqlParser.java new file mode 100644 index 0000000..dbcb83c --- /dev/null +++ b/tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/SqlParser.java @@ -0,0 +1,84 @@ +package com.transferwise.common.entrypoints.tableaccessstatistics; + +import java.time.Duration; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.regex.Matcher; +import java.util.regex.Pattern; +import net.sf.jsqlparser.JSQLParserException; +import net.sf.jsqlparser.parser.CCJSqlParser; +import net.sf.jsqlparser.parser.StringProvider; +import net.sf.jsqlparser.statement.Statements; + +public class SqlParser { + + /* + DATABASE is reserved keyword in sql parser, so having DATABASE() in sql will just throw error. + DATABASE() is however used by some of our internal libraries for MariaDb. + */ + private static final Pattern FUNCTION_REPLACEMENT_PATTERN = Pattern.compile("DATABASE()", + Pattern.LITERAL | Pattern.CASE_INSENSITIVE | Pattern.UNICODE_CASE); + + private static final Pattern ON_CONFLICT_REPLACEMENT_PATTERN = Pattern.compile("on\\s*conflict\\s*\\(.*?\\)", + Pattern.CASE_INSENSITIVE | Pattern.UNICODE_CASE | Pattern.DOTALL); + private final ExecutorService executorService; + + public SqlParser(ExecutorService executorService) { + this.executorService = executorService; + } + + public Statements parse(String sql, Duration timeout) throws JSQLParserException { + sql = replaceFunctions(sql); + sql = replaceOnConflicts(sql); + + CCJSqlParser parser = newParser(sql).withAllowComplexParsing(true); + return parseStatementAsync(parser, timeout); + } + + // Sqlparser 4.6 does not support "on conflict" clause with multiple parameters. + // As a workaround, we change the query to a single parameter clause. + protected String replaceOnConflicts(String sql) { + var matcher = ON_CONFLICT_REPLACEMENT_PATTERN.matcher(sql); + + // 99.99% of sqls don't have it, so let's avoid new string creation for those. + if (matcher.find()) { + matcher.reset(); + return matcher.replaceAll(Matcher.quoteReplacement("on conflict (blah)")); + } + + return sql; + } + + protected String replaceFunctions(String sql) { + var matcher = FUNCTION_REPLACEMENT_PATTERN.matcher(sql); + + // 99.99% of sqls don't have it, so let's avoid new string creation for those. + if (matcher.find()) { + matcher.reset(); + return matcher.replaceAll(Matcher.quoteReplacement("UNSUPPORTED()")); + } + + return sql; + } + + protected Statements parseStatementAsync(CCJSqlParser parser, Duration timeout) throws JSQLParserException { + try { + var future = executorService.submit(parser::Statements); + return future.get(timeout.toMillis(), TimeUnit.MILLISECONDS); + } catch (TimeoutException ex) { + parser.interrupted = true; + throw new JSQLParserException("Time out occurred.", ex); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new JSQLParserException(e); + } catch (Exception ex) { + throw new JSQLParserException(ex); + } + } + + protected CCJSqlParser newParser(String sql) { + return new CCJSqlParser(new StringProvider(sql)); + } + +} diff --git a/tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/SqlParserUtils.java b/tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/SqlParserUtils.java deleted file mode 100644 index 838e21c..0000000 --- a/tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/SqlParserUtils.java +++ /dev/null @@ -1,37 +0,0 @@ -package com.transferwise.common.entrypoints.tableaccessstatistics; - -import java.util.regex.Matcher; -import java.util.regex.Pattern; -import lombok.experimental.UtilityClass; -import net.sf.jsqlparser.parser.CCJSqlParser; -import net.sf.jsqlparser.parser.CCJSqlParserUtil; -import net.sf.jsqlparser.parser.ParseException; -import net.sf.jsqlparser.statement.Statements; - -@UtilityClass -public class SqlParserUtils { - - /* - DATABASE is reserved keyword in sql parser, so having DATABASE() in sql will just throw error. - DATABASE() is however used by some of our internal libraries for MariaDb. - */ - private static final Pattern FUNCTION_REPLACEMENT_PATTERN = Pattern.compile("DATABASE()", - Pattern.LITERAL | Pattern.CASE_INSENSITIVE | Pattern.UNICODE_CASE); - - public Statements parseToStatements(String sql) throws ParseException { - var matcher = FUNCTION_REPLACEMENT_PATTERN.matcher(sql); - - // 99.99% of sqls don't have it, so let's avoid new string creation for those. - if (matcher.find()) { - matcher.reset(); - sql = matcher.replaceAll(Matcher.quoteReplacement("UNSUPPORTED()")); - } - - /* - Don't use `CCJSqlParserUtil.parse`, this has an overhead of launching a new executor service and parsing the sql there. - */ - CCJSqlParser parser = CCJSqlParserUtil.newParser(sql).withAllowComplexParsing(false); - - return parser.Statements(); - } -} diff --git a/tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/TableAccessStatisticsSpyqlListener.java b/tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/TableAccessStatisticsSpyqlListener.java index 346ff09..28f55a6 100644 --- a/tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/TableAccessStatisticsSpyqlListener.java +++ b/tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/TableAccessStatisticsSpyqlListener.java @@ -12,7 +12,9 @@ import com.transferwise.common.context.TwContext; import com.transferwise.common.context.TwContextMetricsTemplate; import com.transferwise.common.entrypoints.EntryPointsMetrics; -import com.transferwise.common.entrypoints.tableaccessstatistics.TableAccessStatisticsSpyqlListener.SqlParseResult.SqlOperation; +import com.transferwise.common.entrypoints.EntryPointsProperties; +import com.transferwise.common.entrypoints.tableaccessstatistics.ParsedQuery.SqlOperation; +import com.transferwise.common.entrypoints.tableaccessstatistics.TasQueryParsingInterceptor.InterceptResult.Decision; import com.transferwise.common.spyql.event.GetConnectionEvent; import com.transferwise.common.spyql.event.StatementExecuteEvent; import com.transferwise.common.spyql.event.StatementExecuteFailureEvent; @@ -21,31 +23,30 @@ import io.micrometer.core.instrument.Gauge; import io.micrometer.core.instrument.MeterRegistry; import io.micrometer.core.instrument.Tag; -import java.util.HashMap; -import java.util.HashSet; +import io.micrometer.core.instrument.binder.cache.CaffeineCacheMetrics; +import java.time.Duration; +import java.time.temporal.ChronoUnit; +import java.util.Collections; import java.util.List; -import java.util.Map; import java.util.Map.Entry; -import java.util.Set; -import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; import java.util.concurrent.TimeUnit; -import lombok.Data; -import lombok.NonNull; -import lombok.experimental.Accessors; import lombok.extern.slf4j.Slf4j; import net.sf.jsqlparser.statement.Statement; -import net.sf.jsqlparser.statement.Statements; import org.apache.commons.lang3.StringUtils; @Slf4j public class TableAccessStatisticsSpyqlListener implements SpyqlDataSourceListener { public static final String GAUGE_SQL_PARSER_RESULT_CACHE_HIT_COUNT = "EntryPoints_Tas_SqlParseResultsCache_hitCount"; + public static final String GAUGE_SQL_PARSER_RESULT_CACHE_MISS_COUNT = "EntryPoints_Tas_SqlParseResultsCache_missCount"; + public static final String GAUGE_SQL_PARSER_RESULT_CACHE_EVICT_COUNT = "EntryPoints_Tas_SqlParseResultsCache_evictCount"; public static final String GAUGE_SQL_PARSER_RESULT_CACHE_HIT_RATIO = "EntryPoints_Tas_SqlParseResultsCache_hitRatio"; public static final String GAUGE_SQL_PARSER_RESULT_CACHE_SIZE = "EntryPoints_Tas_SqlParseResultsCache_size"; public static final String COUNTER_PARSES = "EntryPoints_Tas_Parses"; public static final String COUNTER_FAILED_PARSES = "EntryPoints_Tas_FailedParses"; + public static final String COUNTER_SLOW_PARSES = "EntryPoints_Tas_SlowParses"; public static final String COUNTER_UNCOUNTED_QUERIES = "EntryPoints_Tas_UncountedQueries"; public static final String TIMER_FIRST_TABLE_ACCESS = "EntryPoints_Tas_FirstTableAccess"; public static final String COUNTER_TABLE_ACCESS = "EntryPoints_Tas_TableAccess"; @@ -63,24 +64,47 @@ public class TableAccessStatisticsSpyqlListener implements SpyqlDataSourceListen private final IMeterCache meterCache; - final LoadingCache sqlParseResultsCache; + final LoadingCache sqlParseResultsCache; - public TableAccessStatisticsSpyqlListener(IMeterCache meterCache, Executor executor, String databaseName, - long sqlParserCacheSizeMib) { + private final TasParsedQueryRegistry tasParsedQueryRegistry; + private final SqlParser sqlParser; + private final EntryPointsProperties entryPointsProperties; + private final TasQueryParsingListener tasQueryParsingListener; + private final TasQueryParsingInterceptor tasQueryParsingInterceptor; + + public TableAccessStatisticsSpyqlListener(IMeterCache meterCache, ExecutorService executorService, + TasParsedQueryRegistry tasParsedQueryRegistry, String databaseName, + EntryPointsProperties entryPointsProperties, TasQueryParsingListener tasQueryParsingListener, + TasQueryParsingInterceptor tasQueryParsingInterceptor) { this.databaseName = databaseName; this.dbTag = Tag.of(EntryPointsMetrics.TAG_DATABASE, databaseName); this.meterCache = meterCache; + this.tasParsedQueryRegistry = tasParsedQueryRegistry; + this.sqlParser = new SqlParser(executorService); + this.entryPointsProperties = entryPointsProperties; + this.tasQueryParsingInterceptor = tasQueryParsingInterceptor; + this.tasQueryParsingListener = tasQueryParsingListener; + MeterRegistry meterRegistry = meterCache.getMeterRegistry(); meterRegistry.config().meterFilter(new TasMeterFilter()); - sqlParseResultsCache = Caffeine.newBuilder().maximumWeight(sqlParserCacheSizeMib * MIB).recordStats() - .executor(executor) - .weigher((String k, SqlParseResult v) -> k.length() * 2) + sqlParseResultsCache = Caffeine.newBuilder().maximumWeight(entryPointsProperties.getTas().getSqlParser().getCacheSizeMib() * MIB).recordStats() + .executor(executorService) + .weigher((String k, ParsedQuery v) -> k.length() * 2) .build(sql -> parseSql(sql, TwContext.current())); - Gauge.builder(GAUGE_SQL_PARSER_RESULT_CACHE_SIZE, sqlParseResultsCache::estimatedSize).register(meterRegistry); - Gauge.builder(GAUGE_SQL_PARSER_RESULT_CACHE_HIT_RATIO, () -> sqlParseResultsCache.stats().hitRate()).register(meterRegistry); - Gauge.builder(GAUGE_SQL_PARSER_RESULT_CACHE_HIT_COUNT, () -> sqlParseResultsCache.stats().hitCount()).register(meterRegistry); + new CaffeineCacheMetrics<>(sqlParseResultsCache, "ep-tas-parse-results-" + databaseName, Collections.emptyList()).bindTo(meterRegistry); + + Gauge.builder(GAUGE_SQL_PARSER_RESULT_CACHE_SIZE, sqlParseResultsCache::estimatedSize).tag("database", databaseName) + .register(meterRegistry); + Gauge.builder(GAUGE_SQL_PARSER_RESULT_CACHE_HIT_RATIO, () -> sqlParseResultsCache.stats().hitRate()).tag("database", databaseName) + .register(meterRegistry); + Gauge.builder(GAUGE_SQL_PARSER_RESULT_CACHE_HIT_COUNT, () -> sqlParseResultsCache.stats().hitCount()).tag("database", databaseName) + .register(meterRegistry); + Gauge.builder(GAUGE_SQL_PARSER_RESULT_CACHE_MISS_COUNT, () -> sqlParseResultsCache.stats().missCount()).tag("database", databaseName) + .register(meterRegistry); + Gauge.builder(GAUGE_SQL_PARSER_RESULT_CACHE_EVICT_COUNT, () -> sqlParseResultsCache.stats().evictionCount()).tag("database", databaseName) + .register(meterRegistry); } @Override @@ -88,32 +112,45 @@ public SpyqlConnectionListener onGetConnection(GetConnectionEvent event) { return new ConnectionListener(); } - protected SqlParseResult parseSql(String sql, TwContext context) { - SqlParseResult result = new SqlParseResult(); + protected ParsedQuery parseSql(String sql, TwContext context) { + ParsedQuery result = new ParsedQuery(); + long startTimeMs = System.currentTimeMillis(); try { - Statements stmts = SqlParserUtils.parseToStatements(sql); + var stmts = sqlParser.parse(sql, entryPointsProperties.getTas().getSqlParser().getTimeout()); for (Statement stmt : stmts.getStatements()) { // Intern() makes later equal checks much faster. String opName = getOperationName(stmt).intern(); CustomTablesNamesFinder tablesNamesFinder = new CustomTablesNamesFinder(); - List tableNames = tablesNamesFinder.getTableList(stmt); + List tableNames = null; + try { + tableNames = tablesNamesFinder.getTableList(stmt); + } catch (UnsupportedOperationException e) { + // Some type of statements do not support finding table names. + // For example a statement 'SHOW FULL TABLES IN ...'. + log.debug("Unsupported query '{}'.", sql, e); + } - SqlParseResult.SqlOperation sqlOp = result + ParsedQuery.SqlOperation sqlOp = result .getOperations() - .computeIfAbsent(opName, k -> new SqlParseResult.SqlOperation()); + .computeIfAbsent(opName, k -> new ParsedQuery.SqlOperation()); - for (String tableName : tableNames) { - // Intern() makes later equal checks much faster. - sqlOp.getTableNames().add(tableName.intern()); + if (tableNames != null) { + for (String tableName : tableNames) { + // Intern() makes later equal checks much faster. + sqlOp.getTableNames().add(tableName.intern()); + } } } + meterCache.counter(COUNTER_PARSES, TagsSet.of( EntryPointsMetrics.TAG_DATABASE, databaseName, TwContextMetricsTemplate.TAG_EP_GROUP, context.getGroup(), TwContextMetricsTemplate.TAG_EP_NAME, context.getName(), TwContextMetricsTemplate.TAG_EP_OWNER, context.getOwner() )).increment(); + + tasQueryParsingListener.parsingDone(sql, result, Duration.of(System.currentTimeMillis() - startTimeMs, ChronoUnit.MILLIS)); } catch (Throwable t) { meterCache.counter(COUNTER_FAILED_PARSES, TagsSet.of( EntryPointsMetrics.TAG_DATABASE, databaseName, @@ -121,7 +158,17 @@ protected SqlParseResult parseSql(String sql, TwContext context) { TwContextMetricsTemplate.TAG_EP_NAME, context.getName(), TwContextMetricsTemplate.TAG_EP_OWNER, context.getOwner() )).increment(); - log.debug(t.getMessage(), t); + tasQueryParsingListener.parsingFailed(sql, Duration.of(System.currentTimeMillis() - startTimeMs, ChronoUnit.MILLIS), t); + } finally { + long durationMs = System.currentTimeMillis() - startTimeMs; + if (durationMs > entryPointsProperties.getTas().getSqlParser().getParseDurationWarnThreshold().toMillis()) { + meterCache.counter(COUNTER_SLOW_PARSES, TagsSet.of( + EntryPointsMetrics.TAG_DATABASE, databaseName, + TwContextMetricsTemplate.TAG_EP_GROUP, context.getGroup(), + TwContextMetricsTemplate.TAG_EP_NAME, context.getName(), + TwContextMetricsTemplate.TAG_EP_OWNER, context.getOwner() + )).increment(); + } } return result; @@ -149,9 +196,19 @@ protected void registerSql(String sql, boolean isInTransaction, boolean succeede final Tag inTransactionTag = isInTransaction ? TAG_IN_TRANSACTION_TRUE : TAG_IN_TRANSACTION_FALSE; final Tag successTag = succeeded ? TAG_SUCCESS_TRUE : TAG_SUCCESS_FALSE; - SqlParseResult sqlParseResult = sqlParseResultsCache.get(sql, sqlForCache -> parseSql(sqlForCache, context)); + ParsedQuery parsedQuery = null; - if (sqlParseResult == null) { + var interceptResult = tasQueryParsingInterceptor.intercept(sql); + if (interceptResult.getDecision() == Decision.CONTINUE) { + parsedQuery = tasParsedQueryRegistry.get(sql); + if (parsedQuery == null) { + parsedQuery = sqlParseResultsCache.get(sql, sqlForCache -> parseSql(sqlForCache, context)); + } + } else if (interceptResult.getDecision() == Decision.CUSTOM_PARSED_QUERY) { + parsedQuery = interceptResult.getParsedQuery(); + } + + if (parsedQuery == null) { meterCache.counter(COUNTER_UNCOUNTED_QUERIES, TagsSet.of( EntryPointsMetrics.TAG_DATABASE, databaseName, TwContextMetricsTemplate.TAG_EP_GROUP, context.getGroup(), @@ -162,43 +219,30 @@ protected void registerSql(String sql, boolean isInTransaction, boolean succeede return; } - for (Entry entry : sqlParseResult.operations.entrySet()) { + for (Entry entry : parsedQuery.getOperations().entrySet()) { String opName = entry.getKey(); SqlOperation op = entry.getValue(); - String firstTableName = null; - for (String tableName : op.getTableNames()) { - TagsSet tagsSet = TagsSet.of( - dbTag.getKey(), dbTag.getValue(), - TwContextMetricsTemplate.TAG_EP_GROUP, context.getGroup(), - TwContextMetricsTemplate.TAG_EP_NAME, context.getName(), - TwContextMetricsTemplate.TAG_EP_OWNER, context.getOwner(), - inTransactionTag.getKey(), inTransactionTag.getValue(), - TAG_OPERATION, opName, - successTag.getKey(), successTag.getValue(), - TAG_TABLE, tableName); - - if (firstTableName == null) { - firstTableName = tableName; - meterCache.timer(TIMER_FIRST_TABLE_ACCESS, tagsSet).record(executionTimeNs, TimeUnit.NANOSECONDS); + if (op.getTableNames() != null) { + String firstTableName = null; + for (String tableName : op.getTableNames()) { + TagsSet tagsSet = TagsSet.of( + dbTag.getKey(), dbTag.getValue(), + TwContextMetricsTemplate.TAG_EP_GROUP, context.getGroup(), + TwContextMetricsTemplate.TAG_EP_NAME, context.getName(), + TwContextMetricsTemplate.TAG_EP_OWNER, context.getOwner(), + inTransactionTag.getKey(), inTransactionTag.getValue(), + TAG_OPERATION, opName, + successTag.getKey(), successTag.getValue(), + TAG_TABLE, tableName); + + if (firstTableName == null) { + firstTableName = tableName; + meterCache.timer(TIMER_FIRST_TABLE_ACCESS, tagsSet).record(executionTimeNs, TimeUnit.NANOSECONDS); + } + meterCache.counter(COUNTER_TABLE_ACCESS, tagsSet).increment(); } - meterCache.counter(COUNTER_TABLE_ACCESS, tagsSet).increment(); } } } } - - @Data - @Accessors(chain = true) - static class SqlParseResult { - - @NonNull - private Map operations = new HashMap<>(); - - @Data - @Accessors(chain = true) - static class SqlOperation { - - private Set tableNames = new HashSet<>(); - } - } } diff --git a/tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/TasParsedQueryRegistry.java b/tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/TasParsedQueryRegistry.java new file mode 100644 index 0000000..31483e3 --- /dev/null +++ b/tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/TasParsedQueryRegistry.java @@ -0,0 +1,11 @@ +package com.transferwise.common.entrypoints.tableaccessstatistics; + +/** + * Allows to register queries which are not parseable or very slow to parse. + */ +public interface TasParsedQueryRegistry { + + void register(String sql, ParsedQuery parsedQuery); + + ParsedQuery get(String sql); +} diff --git a/tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/TasQueryParsingInterceptor.java b/tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/TasQueryParsingInterceptor.java new file mode 100644 index 0000000..4d09efc --- /dev/null +++ b/tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/TasQueryParsingInterceptor.java @@ -0,0 +1,36 @@ +package com.transferwise.common.entrypoints.tableaccessstatistics; + +import lombok.Data; +import lombok.experimental.Accessors; + +public interface TasQueryParsingInterceptor { + + InterceptResult intercept(String sql); + + @Data + @Accessors(chain = true) + class InterceptResult { + + private Decision decision = Decision.CONTINUE; + + private ParsedQuery parsedQuery; + + public static InterceptResult doContinue() { + return new InterceptResult().setDecision(Decision.CONTINUE); + } + + public static InterceptResult doSkip() { + return new InterceptResult().setDecision(Decision.SKIP); + } + + public static InterceptResult returnParsedQuery(ParsedQuery parsedQuery) { + return new InterceptResult().setDecision(Decision.CUSTOM_PARSED_QUERY).setParsedQuery(parsedQuery); + } + + enum Decision { + SKIP, + CUSTOM_PARSED_QUERY, + CONTINUE + } + } +} diff --git a/tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/TasQueryParsingListener.java b/tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/TasQueryParsingListener.java new file mode 100644 index 0000000..b243169 --- /dev/null +++ b/tw-entrypoints/src/main/java/com/transferwise/common/entrypoints/tableaccessstatistics/TasQueryParsingListener.java @@ -0,0 +1,10 @@ +package com.transferwise.common.entrypoints.tableaccessstatistics; + +import java.time.Duration; + +public interface TasQueryParsingListener { + + void parsingDone(String sql, ParsedQuery parsedQuery, Duration timeTaken); + + void parsingFailed(String sql, Duration timeTaken, Throwable t); +} diff --git a/tw-entrypoints/src/test/java/com/transferwise/common/entrypoints/tableaccessstatistics/SqlParserUtilsTest.java b/tw-entrypoints/src/test/java/com/transferwise/common/entrypoints/tableaccessstatistics/SqlParserUtilsTest.java index f342b85..d2a8a75 100644 --- a/tw-entrypoints/src/test/java/com/transferwise/common/entrypoints/tableaccessstatistics/SqlParserUtilsTest.java +++ b/tw-entrypoints/src/test/java/com/transferwise/common/entrypoints/tableaccessstatistics/SqlParserUtilsTest.java @@ -1,28 +1,43 @@ package com.transferwise.common.entrypoints.tableaccessstatistics; +import java.time.Duration; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.Executors; import lombok.SneakyThrows; -import net.sf.jsqlparser.parser.ParseException; +import net.sf.jsqlparser.JSQLParserException; import org.junit.jupiter.api.Test; -public class SqlParserUtilsTest { +class SqlParserUtilsTest { @Test @SneakyThrows void testJSqlParser() { List sqls = new ArrayList<>(); + sqls.add("insert into fin_unique_tw_task_key(task_id,key_hash,key) values(?, ?, ?) on conflict (key_hash, key) do nothing"); + // DATABASE is a keyword for sql parser. sqls.add("select DATABASE()"); sqls.add("delete tl from tag_links tl, tags t where t.name=? and tl.type=? and tl.tag_ref=? and tl.tag_id = t.id"); sqls.add("select table_rows from information_schema.tables where table_schema=DATABASE() and table_name = 'tw_task'"); + sqls.add("SHOW FULL TABLES IN fx WHERE TABLE_TYPE NOT LIKE 'VIEW'"); + + // Skipped by SqlFilter, because sqlparser can not handle this. + // sqls.add("SET statement_timeout TO '18000'"); + + sqls.add("insert into fin_unique_tw_task_key(task_id,key_hash,key) values(?, ?, ?) on conflict (key_hash, key) do nothing"); + + var sqlParser = new SqlParser(Executors.newCachedThreadPool()); + for (String sql : sqls) { try { - SqlParserUtils.parseToStatements(sql); - } catch (ParseException e) { + var statements = sqlParser.parse(sql, Duration.ofSeconds(5)); + + statements.getStatements(); + } catch (JSQLParserException e) { throw new RuntimeException("Failed to parse sql '" + sql + "'.", e); } }